RabbitMQ学习笔记(二)SpringAMQP的使用、消息转换器

前言

RabbitMQ学习笔记(一)RabbitMQ部署、5种队列模型

3 SpringAMQP

3.1 介绍

AMQP（Advanced Message Queuing Protocol），是一个应用程序之间传递业务消息的标准高级消息队列协议。基于此协议的客户端与消息中间件可传递消息，并不受不同的开发语言等条件的限制。

SpringAMQP是基于AMQP协议定义的一套RabbitMQ模板，并且利用SpringBoot对其实现了自动装配，使用起来非常方便。SpringAmqp的官方地址：https://spring.io/projects/spring-amqp。

SpringAMQP主要提供了三个功能：

• 自动声明队列、交换机及其绑定关系
• 基于注解的监听器模式，异步接收消息
• 封装了RabbitTemplate工具，用于发送消息

3.2 简单队列模型

• 1）引入依赖

<dependency>
    <groupId>org.springframework.bootgroupId>
    <artifactId>spring-boot-starter-amqpartifactId>
dependency>

• 2）在application.yml文件中配置RabbitMQ

spring:
  rabbitmq:
    host: 192.168.153.128
    port: 5672
    virtual-host: /
    username: rabbitmq
    password: 123321

• 3）利用RabbitTemplate实现消息发送

@Autowired
private RabbitTemplate rabbitTemplate;

@Test
public void testSimpleQueue() {
    // 队列名称
    String queueName = "simple.queue";
    // 消息
    String message = "hello, spring amqp!";
    // 发送消息
    rabbitTemplate.convertAndSend(queueName, message);
}

执行以上单元测试，在RabbitMQ管理页面查看队列中的消息：

• 4）利用RabbitListener实现消息接收

@Component
public class SpringRabbitListener {

    @RabbitListener(queues = "simple.queue")
    public void listenSimpleQueueMessage(String msg) throws InterruptedException {
        System.out.println("spring 消费者接收到消息：【" + msg + "】");
    }
}

• 5）测试结果

3.3 工作队列模型

工作队列模型即让多个消费者绑定到一个队列，共同消费队列中的消息。

• 1）利用RabbitTemplate和循环实现消息批量发送

@Autowired
private RabbitTemplate rabbitTemplate;

@Test
public void testWorkQueue() {
    // 队列名称
    String queueName = "work.queue";
    // 发送消息
    String message = "hello, work queue ";
    for (int i = 0; i < 50; i++) {
        rabbitTemplate.convertAndSend(queueName, message + i);
    }
}

执行以上单元测试，在RabbitMQ管理页面查看队列中的消息：

• 2）利用RabbitListener实现消息接收，模拟工作队列

@Component
public class SpringRabbitListener {

    @RabbitListener(queues = "work.queue")
    public void listenWorkQueueMessage1(String msg) {
        System.out.println("work.queue监听器1：" + msg);
    }

    @RabbitListener(queues = "work.queue")
    public void listenWorkQueueMessage2(String msg) {
        System.out.println("work.queue监听器2：" + msg);
    }
}

• 3）测试结果

3.4 发布/订阅模型

由上图可知，发布/订阅模型包含的角色如下：

• Publisher：生产者，但其消息不再直接发送到队列中，而是发给exchange（交换机）。
• Exchange：交换机。一方面，接收生产者发送的消息。另一方面，决定如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有以下3种类型：
  • Fanout：广播，将消息发送给所有绑定到交换机的队列。
  • Direct：定向，把消息发送符合指定routing key的队列。
  • Topic：通配符，把消息发送给符合routing pattern（路由模式）的队列。
• Consumer：消费者，与其他模型一样。
• Queue：消息队列，与其他模型一样。

需要注意的是，Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！

3.4.1 Fanout广播模型

在Fanout广播模型中，Exchange（交换机）会将消息发送给所有绑定到交换机的队列。

• 1）声明交换机和队列，及其绑定关系

@Configuration
public class FanoutConfig {

    /**
     * 声明交换机
     */
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("star.fanout");
    }

    /**
     * 第1个队列
     */
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");
    }

    /**
     * 绑定队列1和交换机
     */
    @Bean
    public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }

    /**
     * 第2个队列
     */
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }

    /**
     * 绑定队列2和交换机
     */
    @Bean
    public Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }
}

• 2）利用RabbitTemplate发送消息到交换机

@Autowired
private RabbitTemplate rabbitTemplate;

@Test
public void testFanoutExchange() {
    // 队列名称
    String exchangeName = "star.fanout";
    // 消息
    String message = "hello, fanout!";
    rabbitTemplate.convertAndSend(exchangeName, "", message);
}

• 3）利用RabbitListener接收消息

@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String msg) {
    System.out.println("消费者1接收到Fanout消息：【" + msg + "】");
}

@RabbitListener(queues = "fanout.queue2")
public void listenFanoutQueue2(String msg) {
    System.out.println("消费者2接收到Fanout消息：【" + msg + "】");
}

• 4）测试结果

3.4.2 Direct定向模型

在Direct定向模型中，Exchange（交换机）会把消息发送符合指定routing key的队列。

因此，交换机与队列与的绑定，不再是任意绑定，而是要指定一个RoutingKey（路由key）；生产者在向Exchange发送消息时，也必须指定消息的RoutingKey。

只有队列的RoutingKey与消息的RoutingKey完全一致时，才会接收到消息。

• 1）声明交换机和队列，及其绑定关系与RoutingKey

在Fanout广播模型的案例中，使用的是@Bean的方式声明队列和交换机，比较麻烦。Spring还提供了基于注解方式来声明：

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "direct.queue1"),
        exchange = @Exchange(name = "star.direct", type = ExchangeTypes.DIRECT),
        key = {"red", "blue"}
))
public void listenDirectQueue1(String msg){
    System.out.println("消费者1接收到Direct的消息：【" + msg + "】");
}

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "direct.queue2"),
        exchange = @Exchange(name = "star.direct", type = ExchangeTypes.DIRECT),
        key = {"red", "yellow"}
))
public void listenDirectQueue2(String msg){
    System.out.println("消费者2接收到Direct的消息：【" + msg + "】");
}

• 2）利用RabbitTemplate发送消息到交换机

@Autowired
private RabbitTemplate rabbitTemplate;

@Test
public void testDirectExchange() {
    // 队列名称
    String exchangeName = "star.direct";
    // 消息
    String message = "新的风暴已经出现！";
    rabbitTemplate.convertAndSend(exchangeName, "yellow", message);
}

• 3）测试结果

3.4.3 Topic通配符模型

Topic通配符模型和Direct定向模型一样，都是可以根据RoutingKey把消息路由到不同的队列，只不过Topic通配符模型的RoutingKey可以使用通配符！

Routingkey一般都是有一个或多个单词组成，多个单词之间以.分割,而通配符规则如下：

• #：匹配一个或多个词
• *：匹配恰好1个词

例如：

• item.#：能够匹配item.insert.user 或者 item.insert

• item.*：只能匹配item.insert

• 1）声明交换机和队列，及其绑定关系与RoutingKey

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "topic.queue1"),
        exchange = @Exchange(name = "star.topic", type = ExchangeTypes.TOPIC),
        key = {"item.#"}
))
public void listenTopicQueue1(String msg){
    System.out.println("消费者1接收到Topic的消息：【" + msg + "】");
}

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "topic.queue2"),
        exchange = @Exchange(name = "star.topic", type = ExchangeTypes.TOPIC),
        key = {"item.*"}
))
public void listenTopicQueue2(String msg){
    System.out.println("消费者2接收到Topic的消息：【" + msg + "】");
}

• 2）利用RabbitTemplate发送消息到交换机

@Autowired
private RabbitTemplate rabbitTemplate;

@Test
public void testTopicExchange() {
    String exchangeName = "star.topic";
    String message = "新增用户！";
    rabbitTemplate.convertAndSend(exchangeName, "item.insert.user", message);
    String message2 = "新增***！";
    rabbitTemplate.convertAndSend(exchangeName, "item.insert", message2);
}

• 3）测试结果

3.5 消息转换器

在调用RabbitTemplate的convertAndSend()方法时，Spring会把发送的消息序列化为字节发送给MQ，接收消息时还会把字节反序列化为Java对象。

默认情况下Spring采用的序列化方式是JDK序列化。而这种方式存在下列问题：数据体积过大、有安全漏洞、可读性差。

例如执行以下单元测试：

@Test
public void testSendMap() throws InterruptedException {
    Map<String,Object> msg = new HashMap<>();
    msg.put("name", "Jack");
    msg.put("age", 21);
    rabbitTemplate.convertAndSend("simple.queue","", msg);
}

在RabbitMQ管理页面查看消息：

显然，可读性非常差。JDK序列化方式并不好用。而要使可读性更高，可以使用JSON方式来做序列化和反序列化。

• 1）引入依赖

<dependency>
    <groupId>com.fasterxml.jackson.dataformatgroupId>
    <artifactId>jackson-dataformat-xmlartifactId>
    <version>2.9.10version>
dependency>

• 2）配置消息转换器，在启动类中添加一个Bean

@Bean
public MessageConverter jsonMessageConverter(){
    return new Jackson2JsonMessageConverter();
}

• 3）再次测试

…

本节完，更多内容请查阅分类专栏：微服务学习笔记

感兴趣的读者还可以查阅我的另外几个专栏：

• SpringBoot源码解读与原理分析
• MyBatis3源码深度解析
• Redis从入门到精通
• MyBatisPlus详解
• SpringCloud学习笔记