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

本文主要是介绍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.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

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

spring:rabbitmq:host: 192.168.153.128port: 5672virtual-host: /username: rabbitmqpassword: 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 {/*** 声明交换机*/@Beanpublic FanoutExchange fanoutExchange() {return new FanoutExchange("star.fanout");}/*** 第1个队列*/@Beanpublic Queue fanoutQueue1(){return new Queue("fanout.queue1");}/*** 绑定队列1和交换机*/@Beanpublic Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}/*** 第2个队列*/@Beanpublic Queue fanoutQueue2(){return new Queue("fanout.queue2");}/*** 绑定队列2和交换机*/@Beanpublic 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.dataformat</groupId><artifactId>jackson-dataformat-xml</artifactId><version>2.9.10</version>
</dependency>

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

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

• 3）再次测试

…

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

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

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

这篇关于RabbitMQ学习笔记(二)SpringAMQP的使用、消息转换器的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---