微服务架构（6）微服务通信

2017-10-05

一、HTTP同步调用

服务器端

一般采用REST API，调用方始终处于等待返回调用的状态，直到调用完成。

可使用Spring的@RequestParam，@PathVariable，@RequestHeader等注解传入参数，使用@RestController来表明是个Rest接口。

客户端-RestTemplate

客户端调用，可使用Spring的RestTemplate的getForObject、exchange等方法发起请求。

客户端-OkHttp

如在Android开发中，也可以使用Square公司的OkHttp来调用，依赖如下。

<dependency>
  <groupId>com.squareup.okhttp3</groupId>
  <artifactId>okhttp</artifactId>
  <version>3.6.0</version>
</dependency>

@Autowired
private OkHttpClient okHttpClient;

public String get(String name) throws IOException {
Request request = new Request.Builder()
    .get()
    .url(String.format("http://localhost/hello?name=%s", rootUri, name))
    .build();
Response response = okHttpClient.newCall(request).execute();
return response.body().string();

客户端-Retrofit

也可以使用Square公司的Retrofit来链式调用，在客户端中定义服务接口，通过注解的方式来绑定HTTP请求，依赖如下。

<dependency>
  <groupId>com.squareup.retrofit2</groupId>
  <artifactId>retrofit</artifactId>
  <version>2.2.0</version>
</dependency>
<dependency>
  <groupId>com.squareup.retrofit2</groupId>
  <artifactId>converter-scalars</artifactId>
  <version>2.2.0</version>
</dependency>
<dependency>
  <groupId>com.squareup.retrofit2</groupId>
  <artifactId>converter-jackson</artifactId>
  <version>2.2.0</version>
</dependency>

//HelloClient.java
public class HelloClient {

  private HelloService helloService;

  @Autowired
  public HelloClient(Retrofit retrofit) {
    helloService = retrofit.create(HelloService.class);
  }

  public String hello(String name) throws IOException {
    return helloService.hello(name).execute().body();
  }
}

//HelloService.java
public interface HelloService {

  @GET("/hello")
  Call<String> say1(@Query("name") String name);
}

二、RPC同步调用

相关概念

远程过程调用RPC（Remote Procedure Call），基于TCP协议。通信过程在传输层完成（相比HTTP通信在应用层完成），因此更高效、支持更高并发。

流行的RPC框架有Google的gRPC，Facebook的Thrift，Twitter的Finagle，阿里的Dubbo，新浪微博的Motan等。

RPC的通信原理如下，两个端的Stub（存根）用于对数据进行序列化和反序列化，调用方式需要服务端和客户端建立Socket连接来实现二进制数据的交换。

gRPC

gRPC使用Protocol Buffers来编写服务定义，拥有强大的二进制序列化工具集。两者都是Google的开源项目。

『gRPC官网传送门』

调用步骤：

使用Protocol Buffers规范对RPC接口进行定义，生成所需Stub库。
开发RPC服务端，继承Stub代码库中的抽象类完成业务逻辑实现，并且启动RPC服务。
开发RPC客户端，使用Stub代码库构建Stub对象和RPC请求对象，通过Stub对象发送RPC请求，最终获取RPC响应对象。
如需要通过SpringBoot集成gRPC，服务端可通过注解，自动扫描带有该注解的类，并且自动添加到ServerBuilder中。

三、MQ异步调用

并非所有场景都需要同步，客户端调用服务器不需要等待服务器做出响应，可使用异步调用方式。虽然HTTP和RPC通过改造也可实现异步调用，但基于MQ（Message Queue）消息队列的方式更容易实现服务间的异步调用。

JMS和AMQP

接触消息队列时，总是会提到如JMS、AMQP。

JMS（Java Message Service，Java消息服务）

	JMS	AMQP
定义	Java Message Service，Java消息服务	Wire-protocol
跨语言	否	是
跨平台	否	是
Model	两种消息模型：Peer-2-Peer，Pub/sub	五种消息模型：direct exchange，fanout exchange，topic change，headers exchange，system exchange
支持消息类型	TextMessage、MapMessage、BytesMessage、StreamMessage、ObjectMessage、Message （只有消息头和属性）	byte[]，当实际应用时，有复杂的消息，可以将消息序列化后发送。
综合评价	JMS 定义了JAVA API层面的标准；在java体系中，多个client均可以通过JMS进行交互，不需要应用修改代码，但是其对跨平台的支持较差；	AMQP定义了wire-level层的协议标准；天然具有跨平台、跨语言特性。

ActiveMQ实现JMS异步调用

『ActiveMQ』是Java流行的开源消息中间件。

ActiveMQ可支持JMS（Java Message Service，Java消息服务）技术规范，可直接接入Spring框架。SpringBoot也提供ActiveMQ的插件。

1	$ docker run -d -p 8161:8161 -p 61616:61616 -e ACTIVEMQ_ADMIN_LOGIN=admin -e ACTIVEMQ_ADMIN_PASSWORD=admin --name activemq webcenter/activemq

61616为ActiveMQ监听的TCP端口号。

浏览器访问 http://localhost:8161 打开控制台，功能菜单有

Home，查看ActiveMQ基本信息
Queue，查看管理的队列
Topic，查看管理的主题
Subscribes：查看主题的订阅者
Connections，查看客户端的连接信息
Network，查看网络信息
Scheduled，查看运行的定时任务
Send，通过表单向队列或主题发送具体消息

Queue解决消息的“点对点（Peer 2 Peer）”通信，消息从生产者（Producer）发出后，进入ActiveMQ的Queue，然后将消息传送给其中一个消费者（Consumer）。

Topic解决消息的“发布与订阅（Publish-Subscribe）”通信，生产者（Producer）发布到ActiveMQ的Topic上，然后分发给每个订阅者（Subscriber）。

ActiveMQ和SpringBoot集成

1、客户端（Producer）和服务端（Consumer）依赖

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-activemq</artifactId>
</dependency>

2、客户端和服务端配置文件

1
2
3

spring.activemq.broker-url=tcp://localhost:61616
spring.activemq.user=admin
spring.activemq.password=admin

3、服务端代码

@Component
public class HelloServer {

  @JmsListener(destination = "hello-queue")
  public void receive(String message) {
    System.out.println(message);
  }
}

4、客户端代码

//HelloClient.java

@Component
public class HelloClient {

  @Autowired
  private JmsTemplate jmsTemplate;

  public void send(String message) {
    jmsTemplate.convertAndSend("hello-queue", message);
  }
}

//HelloClientApplication.java

@SpringBootApplication
public class HelloClientApplication {

  @Autowired
  private HelloClient helloClient;

  @PostConstruct
  public void init() {
    StopWatch stopWatch = new StopWatch();
    stopWatch.start();

    for (int i = 0; i < 1000; i++) {
      helloClient.send("hello world");
    }

    stopWatch.stop();
    System.out.println("time: " + stopWatch.getTotalTimeMillis() + "ms"); // time: 56414ms
  }

  public static void main(String[] args) {
    SpringApplication.run(HelloClientApplication.class, args).close();
  }
}

5、测试

docker启动ActiveMQ
启动服务端
启动客户端
查看服务端Console日志打印信息
浏览器登录控制台，可看到Queue有如下调用信息。
可在控制台点击“Send to”发送消息，服务端也可接口消息并打印。

RabbitMQ实现AMQP异步调用

『RabbitMQ』不基于JMS规范，采用Erlang语言实现，性能更好。提供众多语言的客户端，能和Spring、SpringBoot进行整合。

1	$ docker run -d -p 15672:15672 -p 5672:5672 -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin --name rabbitmq rabbitmq:3-management

5672为RabbitMQ监听的TCP端口号。Rabbit只有Queue，没有Topic，可通过Exchange与Queue来实现Topic的功能。

浏览器访问 http://localhost:15672 打开控制台，功能菜单有

Overview，查看RabbitMQ基本信息
Connections，查看客户端的连接信息
Channels，查看RabbitMQ的通道
Exchanges：查看RabbitMQ的交换机
Queue，查看RabbitMQ的队列
Admin，管理用户、虚拟主机和策略等数据

RabbitMQ和SpringBoot集成

1、客户端（Producer）和服务端（Consumer）依赖

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2、客户端和服务端配置文件

1
2
3

spring.rabbitmq.addresses=localhost:5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=admin

3、服务端代码

//HelloServerApplication.java

@SpringBootApplication
public class HelloServerApplication {

  @Bean
  public Queue helloQueue() {
    return new Queue("hello-queue");
  }

  @Bean
  public Jackson2JsonMessageConverter messageConverter() {
    return new Jackson2JsonMessageConverter();
  }

  public static void main(String[] args) {
    SpringApplication.run(HelloServerApplication.class, args);
  }
}


//HelloServer.java
@Component
public class HelloServer {

  @RabbitListener(queues = "hello-queue")
  public void receive(String message) {
    System.out.println(message);
  }

  @RabbitListener(queues = "hello-queue")
  public void receive(Pojo pojo) {
    System.out.println(pojo);
  }
}

4、客户端代码

//HelloClient.java

@Component
public class HelloClient {

  @Autowired
  private RabbitTemplate rabbitTemplate;

  public void send(String message) {
    rabbitTemplate.convertAndSend("hello-queue", message);
  }

  public void send(Pojo pojo) {
    rabbitTemplate.convertAndSend("hello-queue", pojo);
  }
}

//HelloClientApplication.java

@SpringBootApplication
public class HelloClientApplication {

	@Autowired
	private HelloClient helloClient;
	
	@Bean
	public Queue helloQueue() {
		return new Queue("hello-queue");
	}
	
	@Bean
	public Jackson2JsonMessageConverter messageConverter() {
		return new Jackson2JsonMessageConverter();
	}
	
	@PostConstruct
	public void init() {
		helloClient.send("hello world");
		Pojo pojo = new Pojo();
		pojo.setFoo("a");
		pojo.setBar(1);
		helloClient.send(pojo);
	}
	
	public static void main(String[] args) {
		SpringApplication.run(HelloClientApplication.class, args).close();
	}
}

5、测试

docker启动RabbitMQ
启动服务端
启动客户端
查看服务端Console日志打印信息
浏览器登录控制台，可看到Queue的overview菜单、queue菜单、connections菜单的截图如下。

Kafka实现异步调用
kafka是Linkedin于2010年12月份开源的消息发布订阅系统,它主要用于处理活跃的流式数据,大数据量的数据处理上。

Kafka的使用方式类似于rabbitMQ。详情参考最下方代码。

四、请求应答模式实现RPC调用

请求应答模式（Request-Reply）

消息不是从生产者到消费者这样的单向流动，而是一个消息闭环。步骤如下

请求者发送请求消息至请求队列中
应答者从请求队列中获取请求消息
消费者发送应答消息至应答队列
请求者从应答队列中获取应答消息

在RabbitMQ中，每个消息分为3个部分：消息头部、消息属性和消息载荷。头部可自定义，属性只有12种，载荷是编码（序列化）后的数据。

消息属性12种中，有correlation_id和replay_to，分别用于设置消息的关联ID和应答队列名称，可用于请求应答模式。本质上也是一种RPC调用。

RabbitMQ简化了请求应答模式，请求消息与应答消息经历了同样的Exchange与Queue，降低了程序的复杂度。

RabbitMQ的RPC和SpringBoot集成

参考下方代码。

『MQ代码传送门』

五、分布式事务

微服务架构下，当服务之间发生调用时，会产生分布式事务难题。JTA（Java Transaction API，Java事务API）等不轻量级，不跨平台，没必要使用JTA达到数据的“强一致性”，可使用Event-Sourcing（事件溯源）来实现数据的“最终一致性”。

Event-Sourcing需要先进行事件记录，然后才能做到事件溯源。事件记录过程中，先操作模型表，再操作事件表。而事件溯源过程中，先操作事件表，再操作模型表。

事件表字段包含：

ID，标识Event ID，唯一。
Event Type，事件类型，增删改查。
Model Name，模型名称
Model ID，模型对象ID
Created Time，事件的具体时间

以如下经典的分布式调用场景，有Foo Service和Bar Service两个服务，运行在不同进程中，Foo Service插入一条Foo对象到Foo Table中，随后通过MQ调用Bar Service插入一条Bar对象到Bar Table中。如Bar Service插入Bar对象异常，需要回滚Foo Service提交的事务。通过MQ和Event-Soucing能够确保数据的最终一致性，分布式事务控制流程图如下：