Spring Cloud集成RabbitMQ的使用

同步 or 异步

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

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

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

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

  • RabbitMQ
  • Kafka
  • ActiveMQ

异步通信特点:

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

异步的常见形态:

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

MQ应用场景:

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

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

RabbitMQ的基本使用

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

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

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

在配置文件中增加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的使用

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

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的使用

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


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 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集成RabbitMQ的使用

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

Spring Cloud集成RabbitMQ的使用

上一篇:springboot学习


下一篇:C语言 指针 p++ / p-- - C语言零基础入门教程