目录
1、什么是SpringAMQP
2、简单队列模型
1、在父工程pom.xml中引入依赖
<!--AMQP依赖，包含RabbitMQ-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>2、首先在生产者端配置MQ地址，在application.yml中添加配置：
spring:
rabbitmq:
host: 192.168.58.131 #rabbitMQ的ip地址
port: 5672 #端口
username: jie #用户名
password: 1234 #密码
virtual-host: / #虚拟主机2、在生产者端编写测试类SpringAmqpTest，并利用RabbitTemplate实现消息发送：
package com.jie.mq.helloworld.spring;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import java.util.HashMap;
import java.util.Map;
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void test() {
//队列名称
String queueName = "simple.queue";
//消息
String message = "hello, spring amqp!";
//发送消息
rabbitTemplate.convertAndSend(queueName, message);
}
}点击运行后可以到浏览器看看。
3、在消费者端配置MQ地址，在application.yml中添加配置（和在生产者端一致）
4、然后在生产者服务新建一个类，代码如下：
package com.jie.mq.listener;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class SpringRabbitListener {
@RabbitListener(queues = "simple.queue")
public void listenSimpleQueue(String msg){
System.out.println("spring 消费者接收到消息 ：【" + msg + "】");
}
}5、启动消费者，查看控制台和浏览器。
3、WorkQueue 工作队列
3.1 模拟WorkQueue，实现一个队列绑定多个消费者
1、消息发送
这次我们循环发送，模拟大量消息堆积现象。
在生产者服务中添加一个测试方法：
/**
* @description:向队列中不停发送消息，模拟消息堆积。
* @author: jie
* @time: 2022/2/22 9:23
*/
@Test
public void test2() throws InterruptedException {
//队列名称
String queueName = "simple.queue";
//消息
String message = "hello, spring amqp!--";
for (int i = 1; i < 50; i++) {
//发送消息
rabbitTemplate.convertAndSend(queueName, message+i);
//休眠
Thread.sleep(20);
}
}2、消息接收
要模拟多个消费者绑定同一个队列，我们在消费者服务的中添加2个新的方法：
@RabbitListener(queues = "simple.queue")
public void listenWorkQueue1(String msg) throws InterruptedException {
System.out.println("spring 消费者1接收到消息 ：【" + msg + "】"+ LocalTime.now());
//休眠
Thread.sleep(20);
}
@RabbitListener(queues = "simple.queue")
public void listenWorkQueue2(String msg) throws InterruptedException {
System.err.println("消费者2........接收到消息：【" + msg + "】" + LocalTime.now());
//休眠
Thread.sleep(200);
}3、测试
启动消费者后，在执行生产者服务中刚刚编写的发送测试方法InterruptedException。
可以看到消费者1很快完成了自己的25条消息。消费者2却在缓慢的处理自己的25条消息。
也就是说消息是平均分配给每个消费者，并没有考虑到消费者的处理能力。这样显然是有问题的。
4、能者多劳
在spring中有一个简单的配置，可以解决这个问题。我们修改消费者服务的application.yml文件，添加配置：
重启消费者，再重新发送消息。
5、小结
4、发布/订阅
发布订阅模式与之前案例的区别就是允许将同一消息发送给多个消费者。实现方式是加入了exchange（交换机）。
发布订阅的模型如图：
可以看到，在订阅模型中，多了一个exchange角色，而且过程略有变化：
4.1 Fanout（广播）
Fanout，英文翻译是扇出，在MQ中叫广播更合适。
1、在消费者中创建一个配置类，声明队列和交换机：
package com.jie.mq.config;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @description:广播交换机配置类
* @author: jie
* @time: 2022/2/22 11:09
*/
@Configuration
public class FanoutConfig {
/**
* @description:创建交换机itcast.fanout
* @author: jie
* @time: 2022/2/12 19:38
*/
@Bean
public FanoutExchange fanoutExchange(){
return new FanoutExchange("itcast.fanout");
}
/**
* @description:创建队列fanout.queuel
* @author: jie
* @time: 2022/2/12 19:40
*/
@Bean
public Queue fanoutQueue1(){
return new Queue("fanout.queuel");
}
/**
* @description:绑定队列fanout.queuel到交换机
* @author: jie
* @time: 2022/2/12 19:41
*/
@Bean
public Binding fanoutBinding1(Queue fanoutQueue1,FanoutExchange fanoutExchange){
return BindingBuilder
.bind(fanoutQueue1)
.to(fanoutExchange);
}
// fanout.queue2
/**
* @description:创建队列fanout.queue2
* @author: jie
* @time: 2022/2/12 19:40
*/
@Bean
public Queue fanoutQueue2(){
return new Queue("fanout.queue2");
}
/**
* @description:绑定队列fanout.queue2到交换机
* @author: jie
* @time: 2022/2/12 19:41
*/
@Bean
public Binding fanoutBinding2(Queue fanoutQueue2,FanoutExchange fanoutExchange){
return BindingBuilder
.bind(fanoutQueue2)
.to(fanoutExchange);
}
}运行消费者然后查看浏览器。
2、消息发送
在生产者服务的测试类中添加测试方法：
/**
* @description:测试交换机发送消息
* @author: jie
* @time: 2022/2/22 11:18
*/
@Test
public void testFanoutExchange() {
//交换机名称
String exchangeName = "itcast.fanout";
//消息
String message = "hello,every one!";
// 发送消息
rabbitTemplate.convertAndSend(exchangeName,"",message);
}3、消息接收
在消费者服务中添加两个方法，作为消费者：
@RabbitListener(queues = "fanout.queuel")
public void listenFanoutQueue1(String msg){
System.out.println("消费者接收到fanout.queuel的消息 ：【" + msg + "】");
}
@RabbitListener(queues = "fanout.queue2")
public void listenFanoutQueue2(String msg){
System.out.println("消费者接收到fanout.queue2的消息 ：【" + msg + "】");
}4、测试
运行生产者消息发送，启动消费者，查看控制台。
5、小结
交换机的作用是什么？
4.2、Direct（路由）
Direct Exchange 会将接收到的消息根据规则路由到指定的Queue，因此称为路由模式（routes）。
在Direct模型下：
-
队列与交换机的绑定，不能是任意绑定了，而是要指定一个
RoutingKey（路由key） -
消息的发送方在 向 Exchange发送消息时，也必须指定消息的
RoutingKey。 -
Exchange不再把消息交给每一个绑定的队列，而是根据消息的
Routing Key进行判断，只有队列的Routingkey与消息的Routing key完全一致，才会接收到消息
1、消息接收
基于@Bean的方式声明队列和交换机比较麻烦，Spring还提供了基于注解方式来声明。
在消费者r中添加两个消费者，同时基于注解来声明队列和交换机：
@RabbitListener(bindings = @QueueBinding(
//队列
value = @Queue(name = "driect.queue1"),
//交换机
exchange = @Exchange(name = "itcast.direct",type = ExchangeTypes.DIRECT),
key = {"red","blue"}
))
public void listenDirectQueue2(String msg){
System.out.println("消费者接收到direct.queue1的消息 ：【" + msg + "】");
}
@RabbitListener(bindings = @QueueBinding(
//队列
value = @Queue(name = "driect.queue2"),
//交换机
exchange = @Exchange(name = "itcast.direct",type = ExchangeTypes.DIRECT),
key = {"red","yellow"}
))
public void listenDirectQueue1(String msg){
System.out.println("消费者接收到direct.queue2的消息 ：【" + msg + "】");
}
2、消息发送
@Test
public void testSendDirectExchange(){
//交换机名称
String exchangeName = "itcast.direct";
//消息
String message = "hello,jie";
// 发送消息
rabbitTemplate.convertAndSend(exchangeName,"blue",message);
}
3、总结
4.3 Topic（通配符）
1、消息发送
在生产者服务的中添加测试方法：
@Test
public void testSendTopicExchange(){
//交换机名称
String exchangeName = "itcast.topic";
//消息
String message = "2022冬奥运在中国开始了！";
// 发送消息
rabbitTemplate.convertAndSend(exchangeName,"china.weather",message);
}2、消息接收
在消费者服务中添加方法：
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "topic.queue1"),
exchange = @Exchange(name = "itcast.topic",type = ExchangeTypes.TOPIC),
key = "china.#"
))
public void listenTopicQueue1(String msg){
System.out.println("消费者接收到topic.queue1的消息 ：【" + msg + "】");
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "topic.queue2"),
exchange = @Exchange(name = "itcast.topic",type = ExchangeTypes.TOPIC),
key = "#.news"
))
public void listenTopicQueue2(String msg){
System.out.println("消费者接收到topic.queue2的消息 ：【" + msg + "】");
}
5、消息转换器
Spring会把你发送的消息序列化为字节发送给MQ，接收消息的时候，还会把字节反序列化为Java对象。
默认情况下Spring采用的序列化方式是JDK序列化。众所周知，JDK序列化存在下列问题：
我们来测试一下。
5.1 测试默认转换器
@Test
public void test6() {
Map<String,Object> msg = new HashMap<>();
msg.put("name","zs");
msg.put("age",24);
rabbitTemplate.convertAndSend("object.queue",msg);
}我们修改消息发送的代码，发送一个Map对象：
发送消息后查看浏览器：
5.2 配置JSON转换器
在生产者和消费者两个服务中都引入依赖：
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>配置消息转换器。
在启动类中添加一个Bean即可：
@Bean
public MessageConverter messageConverter(){
return new Jackson2JsonMessageConverter();
}
查看浏览器
