使用SpringAMQ实现五种消息队列模型
1.普通队列
2.工作队列(WorkQueue)
发布订阅=>根据交换机的不同分为三种
3.订阅模型之Fanout(广播)
4.订阅模型之Direct(路由)
5.订阅模型之Topic(话题)
使用前导:
1.在生产者和消费者项目上分别导入RabbitMQ依赖
<!--AMQP依赖，包含RabbitMQ-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
2.同样在yaml文件上分别配置MQ地址
一、普通队列
特点:生产者直接将消息发送到一个特定的队列，消费者从该队列接收并处理消息。每个消息都会被一个且只有一个消费者消费
实现思路:
1.定义生产者
RabbitTemplate 是一个核心类，它提供了发送和接收消息到RabbitMQ的基本操作
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
//注入rabbitMQ模板
@Autowired
RabbitTemplate rabbitTemplate;
@Test //普通队列
public void sendMessage1(){
//声明队列名称
String queueName="simple.queue";
//设置发送的消息内容
String sendMessage="明天会更好";
//发送消息
rabbitTemplate.convertAndSend(queueName,sendMessage);
}
}
2.定义消费者
@RabbitListener(queues = "simple.queue") 是Spring AMQP框架中的注解，用于声明一个方法作为消息处理器，监听名为"simple.queue"的RabbitMQ队列。
当在类的方法上使用此注解时，Spring AMQP会自动配置并启动一个消息监听容器来监听指定队列。当"simple.queue"中有新的消息到达时，该注解标记的方法会被调用，并将消息体作为参数传递给该方法进行处理。
@Component
public class SpringListener {
//普通队列
@RabbitListener(queues = "simple.queue")
public void ListenerMessage(String MSG){
System.out.println(MSG);
}
}
二、工作队列(WorkQueue)
特点:多个消费者可以从同一个队列中获取消息，但只能被一个消费者消费一旦消费完就会从队列中删除
WorkQueue解决消息预取分配不均:
(例如有一百条消息 两个消费者平分消息 都获得50条消息 但没有考虑到消费者的处理消息能力,消费者1可以一秒处理50条消息 而消费者2一秒只能处理一条消息 导致消费者1处理完了没事干而消费者2迟迟无法完成任务)
解决办法:在yaml文件中设置预取限额为1 处理完后才能继续获取消息 实现能者多劳加快消息的处理效率
实现思路:
1.定义生产者
@Test //WorkQueue模型
public void sendMessage2() throws InterruptedException {
//WorkQueue模型 只能被一个消费者消费
String queueName="simple.queue";
String sendMessage="明天会更好__";
for (int i =0 ;i<50;i++){
rabbitTemplate.convertAndSend(queueName,sendMessage+1);
Thread.sleep(20);
}
}
2.定义消费者
@RabbitListener(queues = "simple.queue")
public void WorkQueues1(String MSG) throws InterruptedException {
System.out.println("消费者1接收到的消息---["+MSG+"]"+ LocalDateTime.now());
Thread.sleep(20);
}
@RabbitListener(queues = "simple.queue")
public void WorkQueues2(String MSG) throws InterruptedException {
System.err.println("消费者2接收到的消息---["+MSG+"]"+ LocalDateTime.now());
Thread.sleep(200);
}
三、订阅模型之Fanout(广播)
特点:生产者将消息发送到交换机（Exchange），而不是直接发送到队列。交换机会根据绑定规则将消息路由与之绑定的队列。
实现思路(使用配置类声明队列和交换机实现):
1.在消费者服务中在配置类声明队列、交换机，将两者绑定
@Configuration
public class FanoutConfig {
// itcast.fanout交换机
@Bean
public FanoutExchange fanoutExchange(){
return new FanoutExchange("itcast.fanout");
}
// fanout.queue1
@Bean
public Queue fanoutQueue1(){
return new Queue("fanout.queue1");
}
// 绑定队列1到交换机
@Bean
public Binding fanoutBinding1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
return BindingBuilder
.bind(fanoutQueue1)
.to(fanoutExchange);
}
// fanout.queue2
@Bean
public Queue fanoutQueue2(){
return new Queue("fanout.queue2");
}
// 绑定队列2到交换机
@Bean
public Binding fanoutBinding2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
return BindingBuilder
.bind(fanoutQueue2)
.to(fanoutExchange);
}
}
2.在消费者服务中编写消费者方法
@Component
public class SpringListener {
//订阅模型之Fanout(广播) 通过配置类配置交换机和队列实现
@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String msg) {
System.out.println("消费者接收到fanout.queue1的消息：【" + msg + "】");
}
@RabbitListener(queues = "fanout.queue2")
public void listenFanoutQueue2(String msg) {
System.out.println("消费者接收到fanout.queue2的消息：【" + msg + "】");
}
}
3.在生产者中编写测试方法发送消息
@Test //订阅模型之Fanout(广播)
public void sendMessage3(){
//交换机名称
String ExchangeName="itcast.fanout";
//发送的消息
String sendMessage="hello everyone!!!";
rabbitTemplate.convertAndSend(ExchangeName,"",sendMessage);
}
五、订阅模型之Topic(话题)
特点:订阅模型之Topic(话题) 与Direct(路由)类似 但是Direct的key只能是多个单词的列表
而topic可以使用通配符且用.划分 #表示0个或多个单词 如图所示
实现思路:
1.利用@RabbitListener声明Exchange、Queue、key
在消费者服务中编写两个消费者方法分别监听定义的两个队列
@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 + "】");
}
}
2.在生产者中编写发送消息的测试方法
@Test //订阅模型之Topic(话题)
public void sendMessage5(){
//交换机名称
String ExchangeName="itcast.topic";
//发送的消息
String sendMessage="hello word!!!";
rabbitTemplate.convertAndSend(ExchangeName,"china.new",sendMessage);
}