目录
SpringAMQP是基于RabbitMQ封装的一套模板,并且还利用SpringBoot对其实现了自动装配,使用起来非常方便。 SpringAmqp的官方地址:Spring AMQP
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、启动消费者,查看控制台和浏览器。
Work queues,也被称为(Task queues),任务模型。简单来说就是让多个消费者绑定到一个队列,共同消费队列中的消息。
当消息处理比较耗时的时候,可能生产消息的速度会远远大于消息的消费速度。长此以往,消息就会堆积越来越多,无法及时处理。 此时就可以使用work 模型,多个消费者共同处理消息处理,速度就能大大提高了
这次我们循环发送,模拟大量消息堆积现象。
在生产者服务中添加一个测试方法:
/**
* @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个新的方法:
@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);
}
启动消费者后,在执行生产者服务中刚刚编写的发送测试方法InterruptedException。
可以看到消费者1很快完成了自己的25条消息。消费者2却在缓慢的处理自己的25条消息。
也就是说消息是平均分配给每个消费者,并没有考虑到消费者的处理能力。这样显然是有问题的。
在spring中有一个简单的配置,可以解决这个问题。我们修改消费者服务的application.yml文件,添加配置:
重启消费者,再重新发送消息。
Work模型的使用:
发布订阅模式与之前案例的区别就是允许将同一消息发送给多个消费者。实现方式是加入了exchange(交换机)。
发布订阅的模型如图:
可以看到,在订阅模型中,多了一个exchange角色,而且过程略有变化:
Exchange(交换机)只负责转发消息,不具备存储消息的能力,因此如果没有任何队列与Exchange绑定,或者没有符合路由规则的队列,那么消息会丢失!
Fanout,英文翻译是扇出,在MQ中叫广播更合适。
在广播模式下,消息发送流程是这样的:
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);
}
}
运行消费者然后查看浏览器。
在生产者服务的测试类中添加测试方法:
/**
* @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);
}
在消费者服务中添加两个方法,作为消费者:
@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 + "】");
}
运行生产者消息发送,启动消费者,查看控制台。
交换机的作用是什么?
声明队列、交换机、绑定关系的Bean是什么?
在Fanout模式中,一条消息,会被所有订阅的队列都消费。但是,在某些场景下,我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。
Direct Exchange 会将接收到的消息根据规则路由到指定的Queue,因此称为路由模式(routes)。
在Direct模型下:
RoutingKey
(路由key)
RoutingKey
。
Routing Key
进行判断,只有队列的Routingkey
与消息的 Routing key
完全一致,才会接收到消息
基于@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 + "】");
}
@Test
public void testSendDirectExchange(){
//交换机名称
String exchangeName = "itcast.direct";
//消息
String message = "hello,jie";
// 发送消息
rabbitTemplate.convertAndSend(exchangeName,"blue",message);
}
描述下Direct交换机与Fanout交换机的差异?
基于@RabbitListener注解声明队列和交换机有哪些常见注解?
Topic
类型的Exchange
与Direct
相比,都是可以根据RoutingKey
把消息路由到不同的队列。只不过Topic
类型Exchange
可以让队列在绑定Routing key
的时候使用通配符!Routingkey
一般都是有一个或多个单词组成,多个单词之间以”.”分割,例如:item.insert
通配符规则:#
:匹配一个或多个词*
:匹配不多不少恰好1个词 举例:item.#
:能够匹配item.spu.insert
或者item.spu
item.*
:只能匹配item.spu
在生产者服务的中添加测试方法:
@Test
public void testSendTopicExchange(){
//交换机名称
String exchangeName = "itcast.topic";
//消息
String message = "2022冬奥运在中国开始了!";
// 发送消息
rabbitTemplate.convertAndSend(exchangeName,"china.weather",message);
}
在消费者服务中添加方法:
@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 + "】");
}
Spring会把你发送的消息序列化为字节发送给MQ,接收消息的时候,还会把字节反序列化为Java对象。
默认情况下Spring采用的序列化方式是JDK序列化。众所周知,JDK序列化存在下列问题:
我们来测试一下。
@Test
public void test6() {
Map<String,Object> msg = new HashMap<>();
msg.put("name","zs");
msg.put("age",24);
rabbitTemplate.convertAndSend("object.queue",msg);
}
我们修改消息发送的代码,发送一个Map对象:
发送消息后查看浏览器:
显然,JDK序列化方式并不合适。我们希望消息体的体积更小、可读性更高,因此可以使用JSON方式来做序列化和反序列化。
在生产者和消费者两个服务中都引入依赖:
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>
配置消息转换器。
在启动类中添加一个Bean即可:
@Bean
public MessageConverter messageConverter(){
return new Jackson2JsonMessageConverter();
}
查看浏览器