Spring Cloud集成RabbitMQ的使用

2022-01-10 02:44:54

同步 or 异步

前言：我们现在有一个用微服务架构模式开发的系统，系统里有一个商品服务和订单服务，且它们都是同步通信的。

目前我们商品服务和订单服务之间的通信方式是同步的，当业务扩大之后，如果还继续使用同步的方式进行服务之间的通信，会使得服务之间的耦合增大。例如我们登录操作可能需要同步调用用户服务、积分服务、短信服务等等，而服务之间可能又依赖别的服务，那么这样一个登录过程就会耗费不少的时间，以致用户的体验降低。

那我们在微服务架构下要如何对服务之间的通信进行解耦呢？这就需要使用到消息中间件了，消息中间件可以帮助我们将同步的通信转化为异步通信，服务之间只需要对消息队列进行消息的发布、订阅即可，从而解耦服务之间的通信依赖。

目前较为主流的消息中间件：

RabbitMQ
Kafka
ActiveMQ

异步通信特点：

客户端请求不会阻塞进程，服务端的响应可以是非即时的

异步的常见形态：

推送通知
请求/异步响应
消息队列

MQ应用场景：

异步处理
流量削峰
日志处理
应用解耦

更多关于消息中间件的描述，可以参考我另一篇文章：

Java消息中间件的概述与JMS规范

RabbitMQ的基本使用

在上文 Spring Cloud Config - 统一配置中心中，已经演示过使用Docker安装RabbitMQ，所以这里就不再浪费篇幅演示了。

直接进入正题，我们以订单服务和商品服务示例，首先在订单服务的项目中，加入mq的依赖：

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

在配置文件中增加RabbitMQ的相关配置项：

到订单服务的项目中，新建一个message包，在该包中创建一个MqReceiver类，我们来看看RabbitMQ的基本操作。代码如下：

package org.zero.springcloud.order.server.message;
 
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
 
/**
 * @program: sell_order
 * @description: 接收消息，即消费者
 * @author: 01
 * @create: 2018-08-21 22:24
 **/
@Slf4j
@Component
public class MqReceiver {
 
    /**
     * 接收消息并打印
     *
     * @param message message
     */
    @RabbitListener(queues = "myQueue")
    public void process(String message) {
        // @RabbitListener注解用于监听RabbitMQ，queues指定监听哪个队列
        log.info(message);
    }
}

因为RabbitMQ上还没有myQueue这个队列，所以我们还得到RabbitMQ的管理界面上，创建这个队列，如下：

然后新建一个测试类，用于发送消息到队列中，代码如下：

package org.zero.springcloud.order.server;
 
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
 
/**
 * @program: sell_order
 * @description: 发送消息，即消息发布者
 * @author: 01
 * @create: 2018-08-21 22:28
 **/
@RunWith(SpringRunner.class)
@SpringBootTest
public class MqSenderTest {
 
    @Autowired
    private AmqpTemplate amqpTemplate;
 
    @Test
    public void send() {
        for (int i = 0; i < 100; i++) {
            amqpTemplate.convertAndSend("myQueue", "第" + i + "条消息");
        }
    }
}

运行该测试类，运行成功后到OrderApplication的控制台上，看看是否接收并打印了接收到的消息。正常情况应如下：

基本的消费者和发布者的代码我们都已经编写过，并且也测试成功了。但有个小问题，我们要监听一个不存在的队列时，需要手动去新建这个队列，感觉每次都手动新建挺麻烦的。有没有办法当队列不存在时，自动创建该队列呢？答案是有的，依旧使用之前的那个注解，只不过这次的参数要换成queuesToDeclare。示例代码如下：

package org.zero.springcloud.order.server.message;
 
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
 
/**
 * @program: sell_order
 * @description: 接收消息，即消费者
 * @author: 01
 * @create: 2018-08-21 22:24
 **/
@Slf4j
@Component
public class MqReceiver {
 
    /**
     * 接收并打印消息
     * 可以当队列不存在时自动创建队列
     *
     * @param message message
     */
    @RabbitListener(queuesToDeclare = @Queue("myQueue"))
    public void process2(String message) {
        // @RabbitListener注解用于监听RabbitMQ，queuesToDeclare可以创建指定的队列
        log.info(message);
    }
}

以上我们通过示例简单的介绍了消息的收发及队列的创建，本小节则介绍一下exchange 的自动绑定方式。当需要自动绑定 exchange 时，我们也可以通过 bindings 参数完成。示例代码如下：

package org.zero.springcloud.order.server.message;
 
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
 
/**
 * @program: sell_order
 * @description: 接收消息，即消费者
 * @author: 01
 * @create: 2018-08-21 22:24
 **/
@Slf4j
@Component
public class MqReceiver {
 
    /**
     * 接收并打印消息
     * 可以当队列不存在时自动创建队列，以及自动绑定指定的Exchange
     * @param message message
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("myQueue"),
            exchange = @Exchange("myExchange")
    ))
    public void process3(String message) {
        // @RabbitListener注解用于监听RabbitMQ，bindings可以创建指定的队列及自动绑定Exchange
        log.info(message);
    }
}

消息分组我们也是可以通过 bindings 参数完成，例如现在有一个数码供应商服务和一个水果供应商服务，它们都监听着同一个订单服务的消息队列。但我希望数码订单的消息被数码供应商服务消费，而水果订单的消息被水果供应商服务消费。所以我们就需要用到消息分组。示例代码如下：

package org.zero.springcloud.order.server.message;
 
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
 
/**
 * @program: sell_order
 * @description: 接收消息，即消费者
 * @author: 01
 * @create: 2018-08-21 22:24
 **/
@Slf4j
@Component
public class MqReceiver {
 
    /**
     * 数码供应商服务 - 接收消息
     *
     * @param message message
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("computerOrder"),
            exchange = @Exchange("myOrder"),
            key = "computer"  // 指定路由的key
    ))
    public void processComputer(String message) {
        log.info("computer message : {}", message);
    }
 
    /**
     * 水果供应商服务 - 接收消息
     *
     * @param message message
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("computerOrder"),
            exchange = @Exchange("myOrder"),
            key = "fruit"  // 指定路由的key
    ))
    public void processFruit(String message) {
        log.info("fruit message : {}", message);
    }
}

测试代码如下，通过指定key进行消息的分组，将消息发送到数码供应商服务：

package org.zero.springcloud.order.server;
 
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
 
/**
 * @program: sell_order
 * @description: 发送消息，即消息发布者
 * @author: 01
 * @create: 2018-08-21 22:28
 **/
@RunWith(SpringRunner.class)
@SpringBootTest
public class MqSenderTest {
 
    @Autowired
    private AmqpTemplate amqpTemplate;
 
    @Test
    public void sendOrder() {
        for (int i = 0; i < 100; i++) {
            // 第一个参数指定队列，第二个参数来指定路由的key，第三个参数指定消息
            amqpTemplate.convertAndSend("myOrder", "computer", "第" + i + "条消息");
        }
    }
}

重启项目后，运行以上测试代码，控制台输出如下，可以看到只有数码供应商服务才能够接收到消息，而水果供应商服务是接收不到的。这就完成了消息分组：

Spring Cloud Stream的使用

Spring Cloud Stream 是一个用来为微服务应用构建消息驱动能力的框架。它可以基于Spring Boot 来创建独立的，可用于生产的Spring 应用程序。他通过使用Spring Integration来连接消息代理中间件以实现消息事件驱动。Spring Cloud Stream 为一些供应商的消息中间件产品提供了个性化的自动化配置实现，引用了发布-订阅、消费组、分区的三个核心概念。目前仅支持RabbitMQ、Kafka。

什么是Spring Integration ？ Integration 集成

企业应用集成（EAI）是集成应用之间数据和服务的一种应用技术。四种集成风格：

? 文件传输：两个系统生成文件，文件的有效负载就是由另一个系统处理的消息。该类风格的例子之一是针对文件轮询目录或FTP目录，并处理该文件。
共享数据库：两个系统查询同一个数据库以获取要传递的数据。一个例子是你部署了两个EAR应用，它们的实体类（JPA、Hibernate等）共用同一个表。
远程过程调用：两个系统都暴露另一个能调用的服务。该类例子有EJB服务，或SOAP和REST服务。
消息：两个系统连接到一个公用的消息系统，互相交换数据，并利用消息调用行为。该风格的例子就是众所周知的中心辐射式的（hub-and-spoke）JMS架构。

Spring Integration作为一种企业级集成框架，遵从现代经典书籍《企业集成模式》，为开发者提供了一种便捷的实现模式。Spring Integration构建在Spring控制反转设计模式之上，抽象了消息源和目标，利用消息传送和消息操作来集成应用环境下的各种组件。消息和集成关注点都被框架处理，所以业务组件能更好地与基础设施隔离，从而降低开发者所要面对的复杂的集成职责。

模型图：

现在我们来看看Spring Cloud Stream的基本使用，到订单服务项目上，增加如下依赖：

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>

然后是在配置文件中，配置rabbitmq的相关信息，只不过我们之前已经配置过了所以不用配置了。

我们来看看如何使用Spring Cloud Stream发送和接收消息，首先创建一个接口，定义input和output方法。代码如下：

package org.zero.springcloud.order.server.message;
 
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
 
public interface StreamClient {
 
    // 接收消息、入口
    @Input("myMessageInput")
    SubscribableChannel input();
 
    // 发送消息、
    @Output("myMessageOutput")
    MessageChannel output();
}

创建一个消息接收者。代码如下：

package org.zero.springcloud.order.server.message;
 
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.stereotype.Component;
 
/**
 * @program: sell_order
 * @description: 消息接收者
 * @author: 01
 * @create: 2018-08-22 22:16
 **/
@Slf4j
@Component
@EnableBinding(StreamClient.class)
public class StreamReceiver {
 
    @StreamListener("myMessageOutput")
    public void process(String message) {
        log.info("message : {}", message);
    }
}

消息发送者，这里作为一个Controller存在。代码如下：

package org.zero.springcloud.order.server.controller;
 
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.zero.springcloud.order.server.message.StreamClient;
 
/**
 * @program: sell_order
 * @description: 消息发送者
 * @author: 01
 * @create: 2018-08-22 22:18
 **/
@RestController
public class SendMessageController {
 
    private final StreamClient streamClient;
 
    @Autowired
    public SendMessageController(StreamClient streamClient) {
        this.streamClient = streamClient;
    }
 
    @GetMapping("/send/msg")
    public void send() {
        for (int i = 0; i < 100; i++) {
            MessageBuilder<String> messageBuilder = MessageBuilder.withPayload("这是第" + i + "条消息");
            streamClient.output().send(messageBuilder.build());
        }
    }
}

因为我们的微服务可能会部署多个实例，若有多个实例需要对消息进行分组，否则所有的服务实例都会接收到相同的消息。在配置文件中，增加如下配置完成消息的分组：

spring:
  ...
  cloud:
    ...
    stream:
      bindings:
        myMessageOutput:
          group: order
...

重启项目，访问http://localhost:9080/send/msg，控制台输出如下：

注：Spring Cloud Stream可以在项目启动的时候自动创建队列，在项目关闭的时候自动删除队列

在实际的开发中，我们一般发送的消息通常会是一个java对象而不是字符串。所以我们来看看如何发送对象，其实和发送字符串几乎是一样的。消息发送者代码如下：

package org.zero.springcloud.order.server.controller;
 
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.zero.springcloud.order.server.dto.OrderDTO;
import org.zero.springcloud.order.server.message.StreamClient;
 
/**
 * @program: sell_order
 * @description: 消息发送者
 * @author: 01
 * @create: 2018-08-22 22:18
 **/
@RestController
public class SendMessageController {
 
    private final StreamClient streamClient;
 
    @Autowired
    public SendMessageController(StreamClient streamClient) {
        this.streamClient = streamClient;
    }
 
    /**
     * 发送OrderDTO对象
     */
    @GetMapping("/send/msg")
    public void send() {
        OrderDTO orderDTO = new OrderDTO();
        orderDTO.setOrderId("123465");
 
        MessageBuilder<OrderDTO> messageBuilder = MessageBuilder.withPayload(orderDTO);
        streamClient.output().send(messageBuilder.build());
    }
}

消息接收者也只需要在方法参数上声明这个对象的类型即可。代码如下：

package org.zero.springcloud.order.server.message;
 
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.stereotype.Component;
import org.zero.springcloud.order.server.dto.OrderDTO;
 
/**
 * @program: sell_order
 * @description: 消息接收者
 * @author: 01
 * @create: 2018-08-22 22:16
 **/
@Slf4j
@Component
@EnableBinding(StreamClient.class)
public class StreamReceiver {
 
    /**
     * 接收OrderDTO对象
     * @param message message
     */
    @StreamListener("myMessageOutput")
    public void process(OrderDTO message) {
        log.info("message : {}", message);
    }
}

另外需要提到的一点是，默认情况下，java对象在消息队列中是以base64编码存在的，我们也都知道base64不可读。为了方便查看堆积在消息队列里的对象数据，我们希望java对象是以json格式的字符串呈现，这样就方便我们人类阅读。至于这个问题，我们只需要在配置文件中，增加一段content-type的配置即可。如下：

spring:
  ...
  cloud:
    ...
    stream:
      bindings:
        myMessageOutput:
          group: order
          content-type: application/json
...

重启项目，访问http://localhost:9080/send/msg，控制台输出如下：

2018-08-22 23:32:33.704  INFO 12436 --- [nio-9080-exec-4] o.z.s.o.server.message.StreamReceiver    
: message : OrderDTO(orderId=123465, buyerName=null, buyerPhone=null, buyerAddress=null, buyerOpenid=null, 
orderAmount=null, orderStatus=null, payStatus=null, createTime=null, updateTime=null, orderDetailList=null)

当我们接收到消息的时候，可能会需要返回一段特定的消息，表示消息已收到之类的。至于这个功能，我们通过@SendTo注解即可完成。代码如下：

package org.zero.springcloud.order.server.message;
 
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Component;
import org.zero.springcloud.order.server.dto.OrderDTO;
 
/**
 * @program: sell_order
 * @description: 消息接收者
 * @author: 01
 * @create: 2018-08-22 22:16
 **/
@Slf4j
@Component
@EnableBinding(StreamClient.class)
public class StreamReceiver {
 
    /**
     * 接收OrderDTO对象
     * @param message message
     */
    @StreamListener("myMessageOutput")
    @SendTo("myMessageInput")
    public String process(OrderDTO message) {
        log.info("message : {}", message);
 
        return "success";
    }
 
    @StreamListener("myMessageInput")
    public void success(String message) {
        log.info("message : {}", message);
    }
}

重启项目，访问http://localhost:9080/send/msg，控制台输出如下：

Spring Cloud Stream 再一次简化了我们在分布式环境下对消息中间件的操作，配置好消息中间件的连接地址及用户密码后，在开发的过程中，我们只需要关注input和output，对消息中间件的操作基本是无感知的。

Spring Cloud集成RabbitMQ的使用

码农公寓

同步 or 异步

RabbitMQ的基本使用

Spring Cloud Stream的使用

相关文章