RPC框架

什么是RPC

RPC是远程调用过程的简写，是一个协议，处于网络通信协议的第五层：会话层，其下就是TCP/IP协议，在建立在其基础上的通信会话协议。RPC定义了交互的模式，而应用程序使用这些模式，来访问其他服务器的方法，并不需要关系具体的网络上的细节。

RPC 的全称是 Remote Procedure Call 是一种进程间通信方式。它允许程序调用另一个地址空间（通常是共享网络的另一台机器上）的过程或函数，而不用程序员显式编码这个远程调用的细节。即无论是调用本地接口/服务的还是远程的接口/服务，本质上编写的调用代码基本相同。

RPC就是从一台机器（客户端）上通过参数传递的方式调用另一台机器（服务器）上的一个函数或方法（可以统称为服务）并得到返回的结果。
RPC会隐藏底层的通讯细节（不需要直接处理Socket通讯或Http通讯）。
客户端发起请求，服务器返回响应（类似于Http的工作方式）RPC在使用形式上像调用本地函数（或方法）一样去调用远程的函数（或方法）

比如两台服务器A，B，一个应用部署在A服务器上，想要调用B服务器上应用提供的函数或者方法，由于不在一个内存空间，不能直接调用，这时候需要通过就可以应用RPC框架的实现来解决。

RPC 会隐藏底层的通讯细节（不需要直接处理Socket通讯或Http通讯）

RPC 是一个请求响应模型。客户端发起请求，服务器返回响应（类似于Http的工作方式）

RPC 在使用形式上像调用本地函数（或方法）一样去调用远程的函数（或方法）。
1
2
3
4
5

RPC需要解决的三个问题

RPC要达到的目标：远程调用时，要能够像本地调用一样方便，让调用者感知不到远程调用的逻辑

Call ID映射。我们怎么告诉远程机器我们要调用哪个函数呢？在本地调用中，函数体是直接通过函数指针来指定的，我们调用具体函数，编译器就自动帮我们调用它相应的函数指针。但是在远程调用中，是无法调用函数指针的，因为两个进程的地址空间是完全不一样。所以，在RPC中，所有的函数都必须有自己的一个ID。这个ID在所有进程中都是唯一确定的。客户端在做远程过程调用时，必须附上这个ID。然后我们还需要在客户端和服务端分别维护一个 {函数 <–> Call ID} 的对应表。两者的表不一定需要完全相同，但相同的函数对应的Call ID必须相同。当客户端需要进行远程调用时，它就查一下这个表，找出相应的Call ID，然后把它传给服务端，服务端也通过查表，来确定客户端需要调用的函数，然后执行相应函数的代码。

序列化和反序列化。客户端怎么把参数值传给远程的函数呢？在本地调用中，我们只需要把参数压到栈里，然后让函数自己去栈里读就行。但是在远程过程调用时，客户端跟服务端是不同的进程，不能通过内存来传递参数。甚至有时候客户端和服务端使用的都不是同一种语言（比如服务端用C++，客户端用Java或者Python）。这时候就需要客户端把参数先转成一个字节流，传给服务端后，再把字节流转成自己能读取的格式。这个过程叫序列化和反序列化。同理，从服务端返回的值也需要序列化反序列化的过程。

网络传输。远程调用往往是基于网络的，客户端和服务端是通过网络连接的。所有的数据都需要通过网络传输，因此就需要有一个网络传输层。网络传输层需要把Call ID和序列化后的参数字节流传给服务端，然后再把序列化后的调用结果传回客户端。只要能完成这两者的，都可以作为传输层使用。因此，它所使用的协议其实是不限的，能完成传输就行。尽管大部分RPC框架都使用TCP协议，但其实UDP也可以，而gRPC干脆就用了HTTP2。Java的Netty也属于这层的东西

为什么我们要用RPC?

RPC 的主要目标是让构建分布式应用更容易，在提供强大的远程调用能力时不损失本地调用的语义简洁性。为实现该目标，RPC 框架需提供一种透明调用机制让使用者不必显式的区分本地调用和远程调用。

RPC框架实现原理

在RPC框架中主要有三个角色：Provider、Consumer和Registry。如下图所示：

节点角色说明：

Server: 暴露服务的服务提供方。

Client: 调用远程服务的服务消费方。

Registry: 服务注册与发现的注册中心。

PRC架构组件
一个基本的RPC架构里面应该至少包含以下4个组件：

1、客户端（Client）:服务调用方（服务消费者）

2、客户端存根（Client Stub）:存放服务端地址信息，将客户端的请求参数数据信息打包成网络消息，再通过网络传输发送给服务端

3、服务端存根（Server Stub）:接收客户端发送过来的请求消息并进行解包，然后再调用本地服务进行处理

4、服务端（Server）:服务的真正提供者

具体调用过程：

1、服务消费者（client客户端）通过调用本地服务的方式调用需要消费的服务；

2、客户端存根（client stub）接收到调用请求后负责将方法、入参等信息序列化（组装）成能够进行网络传输的消息体；

3、客户端存根（client stub）找到远程的服务地址，并且将消息通过网络发送给服务端；

4、服务端存根（server stub）收到消息后进行解码（反序列化操作）；

5、服务端存根（server stub）根据解码结果调用本地的服务进行相关处理；

6、本地服务执行具体业务逻辑并将处理结果返回给服务端存根（server stub）；

7、服务端存根（server stub）将返回结果重新打包成消息（序列化）并通过网络发送至消费方；

8、客户端存根（client stub）接收到消息，并进行解码（反序列化）；

9、服务消费方得到最终结果；

而RPC框架的实现目标则是将上面的第2-10步完好地封装起来，也就是把调用、编码/解码的过程给封装起来，让用户感觉上像调用本地服务一样的调用远程服务。

RPC调用流程

RPC基本流程图：

从RPC的角度看，应该有服务的提供方，即生产者；还有服务的调用方，即消费者。

对消费者来时，在RPC调用过程中，使用第1步、第2步、第3步、第4步是透明的，其他的都是使用RPC框架去封装这些事情。当应用开始调用PRC的方式时，就会去容器中去取Bean对象，所以我们应该首先注册Bean对象到容器中，我们通过Java的动态代理，将代理过程封装到代理对象中，代理对象实现接口，创建实例到容器中。相应的，在调用远程对象的对象方法时，就会调用动态代理中的方法，这就是代理层的作用。

代理对象在获取到请求方法、接口和参数时，就会用序列化层，将这些信息封装成一个请求报文，再让通信层向服务端传送报文的内容，然后就到了生产者这块。

相应的服务必须有个监听器，来监听来自其他服务的请求，一般都会用容器做消息的监听，就会调用对应的Bean对象的方法，去处理响应的请求。当然，RPC框架不会让容器中的每一个框架都会被调用，所以只有注册了的Bean才会被RPC的请求调用到。然后，通过请求中的类、方法、参数，反射调用对应的Bean，拿到结果之后，通过序列化层，封装好结果报文，服务端的通信层将报文反馈给调用方，调用方解析到返回值，动态代理类返回结果，调用结束。

这样，一个完整的RPC调用反馈链条就完成了。

消费者设计

①代理层： 消费者将对应的接口，通过RPC框架的代理来生成一个对象到Spring容器中。代理层将代理接口生成该接口的对象，该对象处理调用时传过来的对象、方法、参数，通过序列化层封装好，调用网络层。

②序列化层： 将请求的参数序列化成报文；将返回的报文反序列化成对象；

③网络层： 将报文与服务端通信；接收返回结果。

生产者设计

①代理层： 一个应用提供服务，必须由一个网络监听的模块，这个模块大多有开源的容器来处理网络上的监听；服务需要注册，只有注册了的服务才可以被调用；注册的服务需要被我们发射调用到，来进行相应的处理。

②序列化层： 就是相应的做请求的反序列化和结果的序列化。

③网络层： 接收客户端报文；将序列化的结果返回给客户端。

RPC模式总结

1.Proxy代理层

 用于对象的代理；对象的反射调用；RPC流程的控制。

 2.Serialize序列化层

 将请求序列化和结果反序列化。

 3.Invoke网络模块

 主要用于网络通信的相关处理。

 4.Container容器组件

 这层主要用于代理层监听网络请求。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

四、服务注册&发现

服务提供者启动后主动向注册中心注册机器ip、port以及提供的服务列表；

服务消费者启动时向注册中心获取服务提供方地址列表，可实现软负载均衡和Failover；

RPC使用了哪些关键技术？

1、动态代理

生成Client Stub（客户端存根）和Server Stub（服务端存根）的时候需要用到Java动态代理技术，可以使用JDK提供的原生的动态代理机制，也可以使用开源的：CGLib代理，Javassist字节码生成技术。

2、序列化和反序列化

在网络中，所有的数据都将会被转化为字节进行传送，所以为了能够使参数对象在网络中进行传输，需要对这些参数进行序列化和反序列化操作。

序列化：把对象转换为字节序列的过程称为对象的序列化，也就是编码的过程。

反序列化：把字节序列恢复为对象的过程称为对象的反序列化，也就是解码的过程。

目前比较高效的开源序列化框架：如Kryo、FastJson和Protobuf等。

3、NIO通信

出于并发性能的考虑，传统的阻塞式 IO 显然不太合适，因此我们需要异步的 IO，即 NIO。Java 提供了 NIO 的解决方案，Java 7 也提供了更优秀的 NIO.2 支持。可以选择Netty或者MINA来解决NIO数据传输的问题。

4、服务注册中心

可选：Redis、Zookeeper、Consul 、Etcd。一般使用ZooKeeper提供服务注册与发现功能，解决单点故障以及分布式部署的问题(注册中心)。

生产者和消费者调用关系：
1.服务容器负责启动，加载，运行服务提供者。
2.服务提供者在启动时，向注册中心注册自己提供的服务。
3.服务消费者在启动时，向注册中心订阅自己所需的服务。
4.注册中心返回服务提供者地址列表给消费者，如果有变更，注册中心将基于长连接推送变更数据给消费者。
5.服务消费者，从提供者地址列表中，基于软负载均衡算法，选一台提供者进行调用，如果调用失败，再选另一台调用。
6.服务消费者和提供者，在内存中累计调用次数和调用时间，定时每分钟发送一次统计数据到监控中心。

如何使用Java连接Kafka构建生产者和消费者

原文链接：https://blog.csdn.net/Xin_101/article/details/126154171
1 缘起
学习消息队列的过程中，先补习了RabbitMQ相关知识，
接着又重温了Kafka相关的知识，
发现，我并没有积累Java原生操作Kafka的文章，
只使用SpringBoot集成过Kafka，
所以，本次是纯Java的方式操作Kafka，
构建生产者和消费者，本地部署Kafka环境，给出测试样例的测试结果，
同时，讲解部分通用的参数，
及给出通过命令行启动生产者和消费者的测试样例，
分享如下，帮助读者学习Kafka基础操作。

2 环境准备
下载kafka：https://download.csdn.net/download/Xin_101/19787459

2.1 启动zookeeper

bin/zookeeper-server-start.sh config/zookeeper.properties

2.2 启动kafka

bin/kafka-server-start.sh config/server.properties

2.3 新建topic

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic my-topic

2.4 依赖

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.2.0</version>
</dependency>
1
2
3
4
5

3 Kafka通用配置
Kafka的生产者和消费者参数比较多，这里仅列出一些测试用的参数，
区分生产者和消费者。

3.1 生产者配置参数

3.2 消费者配置参数

3.3 Kafka通用参数封装
由于参数众多，这里封装了一个Kafka通用参数类，给了默认值，
本地测试，直接使用默认参数，
同时给出了有参构造器，自定义参数，
代码样例如下。

package com.monkey.java_study.mq.kafka;

import java.util.Collection;
import java.util.Collections;

/**
 * Kafka通用配置.
 *
 * @author xindaqi
 * @since 2022-08-03 9:49
 */
public class KafkaCommonProperties {

    /**
     * Kafka主机
     */
    private String kafkaHost = "192.168.211.129:9092";

    /**
     * 生产者：要求leader请求结束前收到的确认次数，来控制发送数据的持久化
     * 消息确认：
     * 0：生产者不等待服务器确认，此时retry参数不生效
     * 1：leader写入记录到log，不会等待follower的确认即向生产者发送通知
     * all：leader等待所有副本通知，然后向生产者发送通知，保证所有数据落盘到所有副本，功能同设置为-1
     */
    private String ack = "all";

    /**
     * 生产者重试次数
     */
    private Integer retryTimes = 1;

    /**
     * 生产者：向同一分区发送打包发送的数据量，单位：bytes，默认16384bytes=16K
     */
    private Integer batchSize = 16384;

    /**
     * 生产者：批量发送消息的间隔时间（延迟时间），单位：毫秒
     */
    private Integer lingerMs = 1;

    /**
     * 生产者：可以使用的最大缓存空间，单位：bytes，默认33554432bytes=32M.
     */
    private Integer bufferMemory = 33554432;

    /**
     * 生产者：键编码器
     */
    private String keyEncoder = "org.apache.kafka.common.serialization.StringSerializer";

    /**
     * 生产者：值编码器
     */
    private String valueEncoder = "org.apache.kafka.common.serialization.StringSerializer";

    /**
     * 消费者：消费topic的组ID
     */
    private String groupId = "my-group-id";

    /**
     * 消费者：后台定期提交offset
     */
    private String autoCommit = "true";

    /**
     * 消费者提交offset的时间间隔：单位：毫秒，当enable.auto.commit为true时生效
     */
    private String autoCommitIntervalMs = "1000";

    /**
     * 消费者：键解码器
     */
    private String keyDecoder = "org.apache.kafka.common.serialization.StringDeserializer";

    /**
     * 消费者：值解码器
     */
    private String valueDecoder = "org.apache.kafka.common.serialization.StringDeserializer";

    /**
     * 消费者：重启后配置offset
     * earliest：消费者恢复到当前topic最早的offset
     * latest：消费者从最新的offset开始消费
     * none：如果消费者组没找到之前的offset抛出异常
     * 其他任何值都会抛出异常
     */
    private String autoOffsetReset = "latest";

    /**
     * TOPIC
     */
    private Collection<String> topic = Collections.singleton("my-topic");

    public KafkaCommonProperties() {

    }

    public KafkaCommonProperties(String kafkaHost, String ack, Integer retryTimes, Integer batchSize, Integer lingerMs, Integer bufferMemory, String keyEncoder, String valueEncoder, String groupId, String autoCommit, String autoCommitIntervalMs, String keyDecoder, String valueDecoder, String autoOffsetReset, Collection<String> topic) {
        this.kafkaHost = kafkaHost;
        this.ack = ack;
        this.retryTimes = retryTimes;
        this.batchSize = batchSize;
        this.lingerMs = lingerMs;
        this.bufferMemory = bufferMemory;
        this.keyEncoder = keyEncoder;
        this.valueEncoder = valueEncoder;
        this.groupId = groupId;
        this.autoCommit = autoCommit;
        this.autoCommitIntervalMs = autoCommitIntervalMs;
        this.keyDecoder = keyDecoder;
        this.valueDecoder = valueDecoder;
        this.autoOffsetReset = autoOffsetReset;
        this.topic = topic;
    }
// 省略setter和getter及toString()  
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120

4 Code实践
4.1 生产者
构建Kafka数据生产者，
测试样例的配置有：Kafka broker地址，消息确认，重试，批量发送数据，数据键和值的编码器，
重写Callback实现异步生产数据。

4.1.1 生产数据

package com.monkey.java_study.mq.kafka;

import org.apache.kafka.clients.producer.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Objects;
import java.util.Properties;

/**
 * Kafka生产者.
 *
 * @author xindaqi
 * @since 2022-08-02 9:59
 */
public class KafkaProducerTest {

    private static final Logger logger = LoggerFactory.getLogger(KafkaProducerTest.class);

    public static KafkaProducer<String, String> getDefaultKafkaProducer(KafkaCommonProperties kafkaCommonProperties) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaCommonProperties.getKafkaHost());
        properties.put(ProducerConfig.ACKS_CONFIG, kafkaCommonProperties.getAck());
        properties.put(ProducerConfig.RETRIES_CONFIG, kafkaCommonProperties.getRetryTimes());
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, kafkaCommonProperties.getBatchSize());
        properties.put(ProducerConfig.LINGER_MS_CONFIG, kafkaCommonProperties.getLingerMs());
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, kafkaCommonProperties.getBufferMemory());
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, kafkaCommonProperties.getKeyEncoder());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, kafkaCommonProperties.getValueEncoder());
        return new KafkaProducer<>(properties);
    }

    static class MyProducerCallback implements Callback {

        @Override
        public void onCompletion(RecordMetadata metadata, Exception exception) {
            if (Objects.nonNull(exception)) {
                logger.error(">>>>>>>>>>Producer生产消息异常：", exception);
            }
            if (Objects.nonNull(metadata)) {
                logger.info(">>>>>>>>>>Producer生产消息：metadata：{}，partition:{}, offset：{}", metadata, metadata.partition(), metadata.offset());
            }
        }
    }

    public static void main(String[] args) {
        KafkaCommonProperties kafkaCommonProperties = new KafkaCommonProperties();
        KafkaProducer<String, String> producer = getDefaultKafkaProducer(kafkaCommonProperties);
        String message = "hello world ";
        try {
            for (int i = 0; i < 10; i++) {
                // 异步写入数据
                String topic = kafkaCommonProperties.getTopic().toArray()[0].toString();
                ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, message + i);
                producer.send(producerRecord, new MyProducerCallback());
            }
        } catch (Exception ex) {
            logger.error(">>>>>>>>生产数据异常：", ex);
            throw new RuntimeException(ex);
        } finally {
            producer.close();
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

4.1.2 开启生产者
生产者开启后，控制台输出生产者配置信息，如下图所示，其中，
acks在代码中配置为all，而运行日志中acks为-1，所以，acks的all与-1是同种功能。

生产者生产数据是通过异步的方式，控制台日志如下图所示，
由图可知，生产数据的线程为：kafka-producer-network-thread。

4.2 消费者
Kafka消费者通过groupId消费指定topic的，
以groupId区分不同的消费者，即不同的groupId消费相同的topic，对于topic而言，就是不同的消费者，
同时，消费者需要记录消费到的offset，以便下次启动时定位到具体的位置，消费消息。
这里，配置的offset策略为：latest，即每次重启消费者时，从最新的offset开始消费（上次记录的offset之后的一个，如果上次消费没有记录，则从当前offset之后开始消费）。
offset的重置这样理解：
当前topic写入数据有4条，offset从0到3，
如果，offset重设为earliest，则每次重启消费者，offset都会从0开始消费数据；
如果，offset重设为latest，则，每次消费从上次消费的offset下一个开始消费，如果上次消费的offset为3，则，重启后， 从4开始消费数据。

4.2.1 消费数据

package com.monkey.java_study.mq.kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Properties;

/**
 * Kafka消费者.
 *
 * @author xindaqi
 * @since 2022-08-02 9:59
 */
public class KafkaConsumerTest {

    private static final Logger logger = LoggerFactory.getLogger(KafkaConsumerTest.class);

    public static KafkaConsumer<String, String> getDefaultKafkaConsumer(KafkaCommonProperties kafkaCommonProperties) {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaCommonProperties.getKafkaHost());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaCommonProperties.getGroupId());
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, kafkaCommonProperties.getAutoCommit());
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, kafkaCommonProperties.getAutoCommitIntervalMs());
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaCommonProperties.getAutoOffsetReset());
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, kafkaCommonProperties.getKeyDecoder());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, kafkaCommonProperties.getValueDecoder());
        return new KafkaConsumer<>(properties);
    }

    public static void main(String[] args) {
        try {
            KafkaCommonProperties kafkaCommonProperties = new KafkaCommonProperties();
            KafkaConsumer<String, String> consumer = getDefaultKafkaConsumer(kafkaCommonProperties);
            consumer.subscribe(kafkaCommonProperties.getTopic());
            while (Boolean.TRUE) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
                    logger.info(">>>>>>>>Consumer offset:{}, value:{}", record.offset(), record.value());
                }
            }
        } catch (Exception ex) {
            throw new RuntimeException(ex);
        }
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

4.2.2 开启消费者
开启消费者后，控制台输出消费者的配置参数，如下图所示。

同时输出的还有（如下图所示），下次将要消费的offset：41。

消费信息日志如下图所示，
由图可知，从offset 41开始消费。

Eureka服务提供者和消费者

Eureka 服务提供者

创建了注册中心之后，我们就可以创建提供服务的客户端工程了，然后将其注册到注册中心。如果不知道怎么创建注册中心的，请看Spring Cloud系列。
1、首先创建一个Spring Boot项目，添加以下依赖

   <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.1.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

    <properties>
        <java.version>1.8</java.version>
        <spring-cloud.version>Hoxton.SR6</spring-cloud.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
        </dependency>
    </dependencies>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>${spring-cloud.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

2、在启动类添加@EnableDiscoveryClient注解，在Hoxton版本已经可以不用加这个注解

3、在配置文件进行eureka客户端的配置

spring.application.name= eureka-client-service

server.port=8888

#指定服务注册中心地址
eureka.client.serviceUrl.defaultZone=http://localhost:2002/eureka/

# 采用IP注册
eureka.instance.preferIpAddress=true
1
2
3
4
5
6
7
8
9

提示：6666端口号不可用，不要用

启动服务，我们可以看到控制台中有输出注册信息的日志：

2020-07-16 15:29:59.960  INFO 44920 --- [nfoReplicator-0] com.netflix.discovery.DiscoveryClient    : DiscoveryClient_EUREKA-CLIENT-SERVICE/eureka-client-service:192.168.137.1:8888- registration status: 204
1

启动该工程后，再次访问eureka页面，就可以看到我们的client客户端实例了。

至此，一个 Eureka 服务提供者就创建完成了，我们就可以在这个项目上创建接口，让消费者进行消费了

下面是我在服务提供者工程创建的hello接口

@RestController
public class HelloController {
    @GetMapping("/hello")
    public String hello() {
        return "hello";
    }
}
1
2
3
4
5
6
7

Eureka 服务消费者

当我们在一个客户端上提供了一个供其他的客户端使用的服务后，其他客户端如何消费呢。下面我们就来编写一个服务的消费者，来消费一个客户端提供的服务。

首先，创建一个spring boot项目，和服务提供方，添加相应的依赖和配置（将项目名和端口号改一下就行了），将项目注册到和服务提供方相同的注册中心上。

server.port=8890

spring.application.name= eureka-consumer-service

#指定服务注册中心地址
eureka.client.serviceUrl.defaultZone=http://localhost:2002/eureka/```

1
2
3
4
5
6
7

注：最重要的就是将消费者工程和提供者工程都注册到同一个注册中心

两个客户端之间的通信访问，我们使用RestTemplate ，RestTemplate 是 Spring 提供的用于访问 Rest 服务的客户端，RestTemplate 提供了多种便捷访问远程 Http 服务的方法，能够大大提高客户端的编写效率。

//先配置一个RestTemplate来使用

@Configuration
public class RestConfiguration {
    @Bean
    public RestTemplate restTemplate() {
        return new RestTemplate();
    }
}
1
2
3
4
5
6
7

调用服务，我们只需要通过RestTemplate 进行调用即可。调用服务，我们可以使用两种方式，一种是通过url直接调用，一种是通过eureka调用

下面我们就在消费者客户端调用上面服务提供方的hello接口
1、url直接调用

如下：

    @Test
    void contextLoads() {
        String result = restTemplate.getForObject("http://localhost:8888/hello", String.class);
        System.out.println(result);
    }

1
2
3
4
5
6

在消费者客户端调用上面测试方法，成功输出hello

2、通过eureka调用服务
在同一个注册中心中调用服务，我们不需要知道服务提供者的url地址具体是多少，只需要知道服务提供者注册到eureka的名称即可。注册到eureka的名称一般就是spring.application.name

使用eureka调用服务，我们需要在RestTemplate的bean上添加 @LoadBalanced 注解（不添加不能通过服务名调用），这个注解会自动构造 LoadBalancerClient 接口的实现类并注册到 Spring 容器中，然后负载均衡地调用我们的服务实例

所以，我们要修改一下RestTemplate

   //配置RestTemplate 
   @Bean
   @LoadBalanced
    public RestTemplate restTemplate() {
        return new RestTemplate();
    }

1
2
3
4
5
6
7

//服务调用

@Test
void contextLoads() {
    String result = restTemplate.getForObject("http://EUREKA-CLIENT-SERVICE/hello", String.class);
    System.out.println(result);
}

1
2
3
4
5
6