Spring Cloud集成RabbitMQ的使用

曾经终败给现在 2022-04-15 06:25 582阅读 0赞

### 同步 or 异步 ###

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

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

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

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

*  RabbitMQ
 *  Kafka
 *  ActiveMQ

异步通信特点：

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

异步的常见形态：

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

MQ应用场景：

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

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

*  [Java消息中间件的概述与JMS规范][Java_JMS]

--------------------

### RabbitMQ的基本使用（上） ###

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

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

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

在配置文件中增加RabbitMQ的相关配置项：  
![Spring Cloud 集成 RabbitMQ][Spring Cloud _ 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的管理界面上，创建这个队列，如下：  
![Spring Cloud 集成 RabbitMQ][Spring Cloud _ RabbitMQ 1]

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

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的控制台上，看看是否接收并打印了接收到的消息。正常情况应如下：  
![Spring Cloud 集成 RabbitMQ][Spring Cloud _ RabbitMQ 2]

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

--------------------

### RabbitMQ的基本使用（下） ###

以上我们通过示例简单的介绍了消息的收发及队列的创建，本小节则介绍一下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 集成 RabbitMQ][Spring Cloud _ RabbitMQ 3]

--------------------

### Spring Cloud Stream的使用（上） ###

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

什么是Spring Integration ？ Integration 集成

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

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

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

模型图：  
![Spring Cloud 集成 RabbitMQ][Spring Cloud _ RabbitMQ 4]

现在我们来看看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();
    }

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

消息发送者，这里作为一个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 集成 RabbitMQ][Spring Cloud _ RabbitMQ 5]

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

--------------------

### 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 集成 RabbitMQ][Spring Cloud _ RabbitMQ 6]

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

转载出处：[http://blog.51cto.com/zero01/2173288][http_blog.51cto.com_zero01_2173288]

[Java_JMS]: http://blog.51cto.com/zero01/2120351
[Spring Cloud Config -]: http://blog.51cto.com/zero01/2171735
[Spring Cloud _ RabbitMQ]: /images/20220415/f811f617f6044279be0e51c15513c4d3.png
[Spring Cloud _ RabbitMQ 1]: /images/20220415/3db94b509f734abba8394af1e19b13a8.png
[Spring Cloud _ RabbitMQ 2]: /images/20220415/6c70c956fe7a4730ae1682b5f9098102.png
[Spring Cloud _ RabbitMQ 3]: /images/20220415/01aa78c95f00449da2a7c9ddb5adba00.png
[Spring Cloud _ RabbitMQ 4]: /images/20220415/ac254b7431ad412d8359d230c0d03475.png
[Spring Cloud _ RabbitMQ 5]: /images/20220415/75de72e818e741b886d711ea7f2cf794.png
[Spring Cloud _ RabbitMQ 6]: /images/20220415/257de970a4bc427c87470f8f716e6c75.png
[http_blog.51cto.com_zero01_2173288]: http://blog.51cto.com/zero01/2173288