一、rabbitmq的搭建
centos7搭建rabbitmq:centos7安装rabbitmq_java-zh的博客-CSDN博客
二、在什么情况下选择rabbitMQ
常见的四种MQ比较
| 特 性 | ActiveMQ | RabbitMQ | RocketMQ | Kafka |
| 语言 | Java | Erlang | Java | Scala |
| 单机吞吐 | 万 | 万 | 十万 | 十万 |
| 时效性 | ms | us | ms | ms(以内) |
| 可用性 | 高（主从架构) | 高（主从架构) | 非常高 (分布式架构) | 非常高 （分布式架构） |
| 功能特性 | 成熟的产品，在很多公司得到应用；有较多的文档；各种协议支持较好 | 基于erlang开发，所以并发能力很强，性能极其好，延时很低;管理界面较丰富 | MQ功能比较完备，扩展性佳 | 只支持主要的MQ功能，像一些消息查询，消息回溯等功能没有提供，毕竟是为大数据准备的，在大数据领域应用广。 |
三、案例(rabbitMQ常见的五种模式讲解)
3.1 导包
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.4.1</version>
<relativePath/>
</parent>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
<version>3.0.4</version>
</dependency>
</dependencies>yml初始配置
spring:
rabbitmq:
#如果是集群，用,隔开
host: 192.168.139.128
#端口，不能写默认端口15672
port: 5672
connection-timeout: 15000
username: zhonghui
password: 123456
virtual-host: /
#none值是禁用发布确认模式，是默认值
#correlated值是发布消息成功到交换器后会触发回调方法
#simple有两种效果，一种是和correlated值一样会触发回调方法,
#另一种是发布消息成功后使用rabbitTemplate调用waitForConfirms或waitForConfirmsOrDie方法等待broker节点返回发送结果，根据返回结果来判定下一步逻辑,
#要注意的点是waitForConfrimsOrDie方法如果返回false则会关闭Channel,则接下来无法发送消息到broker
publisher-confirm-type: correlated
#消息回退确认机制
publisher-returns: true
listener:
simple:
acknowledge-mode: auto3.2 简单模式
- P:生产者
- C:消费者
Producer
@RestController
@AllArgsConstructor
@RequestMapping("/test/mq")
public class SimpleQueueProducer {
private RabbitTemplate rabbitTemplate;
// 发送到的队列名称
public static final String AMQP_SIMPLE_QUEUE = "amqp.simple.queue";
@GetMapping("/simple/{context}")
public void sendMessage(@PathVariable String context) {
for (int i = 0; i < 10; i++) {
rabbitTemplate.convertAndSend(AMQP_SIMPLE_QUEUE, context + i);
}
}
}Consumer
@Component
@Slf4j
public class SimpleQueueConsumer {
@RabbitListener(queuesToDeclare = @Queue(name = SimpleQueueProducer.AMQP_SIMPLE_QUEUE))
public void consumerSimpleMessage(String context){
// 通过Message解析对象消息
log.info("简单模式内容为:{}",context);
}
}结果
2023-06-20 20:01:43.198 INFO 10440 --- [ntContainer#3-1] c.m.m.controller.SimpleQueueConsumer : 简单模式内容为:mq的简单模式已经完成1
2023-06-20 20:01:43.198 INFO 10440 --- [ntContainer#3-9] c.m.m.controller.SimpleQueueConsumer : 简单模式内容为:mq的简单模式已经完成0
2023-06-20 20:01:43.198 INFO 10440 --- [tContainer#3-10] c.m.m.controller.SimpleQueueConsumer : 简单模式内容为:mq的简单模式已经完成3
2023-06-20 20:01:43.266 INFO 10440 --- [ntContainer#3-8] c.m.m.controller.SimpleQueueConsumer : 简单模式内容为:mq的简单模式已经完成2
2023-06-20 20:01:43.266 INFO 10440 --- [ntContainer#3-7] c.m.m.controller.SimpleQueueConsumer : 简单模式内容为:mq的简单模式已经完成4
2023-06-20 20:01:43.266 INFO 10440 --- [ntContainer#3-6] c.m.m.controller.SimpleQueueConsumer : 简单模式内容为:mq的简单模式已经完成5
2023-06-20 20:01:43.268 INFO 10440 --- [ntContainer#3-4] c.m.m.controller.SimpleQueueConsumer : 简单模式内容为:mq的简单模式已经完成7
2023-06-20 20:01:43.268 INFO 10440 --- [ntContainer#3-3] c.m.m.controller.SimpleQueueConsumer : 简单模式内容为:mq的简单模式已经完成6
2023-06-20 20:01:43.268 INFO 10440 --- [ntContainer#3-2] c.m.m.controller.SimpleQueueConsumer : 简单模式内容为:mq的简单模式已经完成9
2023-06-20 20:01:43.268 INFO 10440 --- [ntContainer#3-5] c.m.m.controller.SimpleQueueConsumer : 简单模式内容为:mq的简单模式已经完成8
注解含义
- @RabbitListener:用于类和方法上
- @queuesToDeclare:将队列绑到默认交换机上
- @Queue:队列注解，value为队列名称
3.3 Work模式
Producer
@RestController
@Slf4j
@AllArgsConstructor
@RequestMapping("/test/mq")
public class WorkProducer {
private RabbitTemplate rabbitTemplate;
// 发送到的队列名称
public static final String AMQP_WORK_QUEUE = "amqp.work.queue";
@GetMapping("/work/{context}")
public void sendMessage(@PathVariable String context) {
for (int i = 0; i < 10; i++) {
rabbitTemplate.convertAndSend(AMQP_WORK_QUEUE,context+i);
}
}
}Consumer
@Component
@Slf4j
public class WorkConsumer {
@RabbitListener(queuesToDeclare = @Queue(value = WorkProducer.AMQP_WORK_QUEUE))
public void consumerSimpleMessage(String context){
try {
// 假设业务逻辑执行了一秒
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
log.info("work1模式内容为:{}",context);
}
}@Component
@Slf4j
public class WorkConsumer2 {
@RabbitListener(queuesToDeclare = @Queue(value = WorkProducer.AMQP_WORK_QUEUE))
public void consumerSimpleMessage2(String context){
log.info("work2模式内容为:{}",context);
}
}结果
2023-06-21 09:06:07.075 INFO 21492 --- [ntContainer#5-1] com.mq.mqcloud.controller.WorkConsumer2 : work2模式内容为:mq的work模式已经完成1
2023-06-21 09:06:07.076 INFO 21492 --- [ntContainer#5-1] com.mq.mqcloud.controller.WorkConsumer2 : work2模式内容为:mq的work模式已经完成3
2023-06-21 09:06:07.076 INFO 21492 --- [ntContainer#5-1] com.mq.mqcloud.controller.WorkConsumer2 : work2模式内容为:mq的work模式已经完成5
2023-06-21 09:06:07.076 INFO 21492 --- [ntContainer#5-1] com.mq.mqcloud.controller.WorkConsumer2 : work2模式内容为:mq的work模式已经完成7
2023-06-21 09:06:07.077 INFO 21492 --- [ntContainer#5-1] com.mq.mqcloud.controller.WorkConsumer2 : work2模式内容为:mq的work模式已经完成9
2023-06-21 09:06:08.053 INFO 21492 --- [ntContainer#4-1] com.mq.mqcloud.controller.WorkConsumer : work1模式内容为:mq的work模式已经完成0
2023-06-21 09:06:09.054 INFO 21492 --- [ntContainer#4-1] com.mq.mqcloud.controller.WorkConsumer : work1模式内容为:mq的work模式已经完成2
2023-06-21 09:06:10.055 INFO 21492 --- [ntContainer#4-1] com.mq.mqcloud.controller.WorkConsumer : work1模式内容为:mq的work模式已经完成4
2023-06-21 09:06:11.057 INFO 21492 --- [ntContainer#4-1] com.mq.mqcloud.controller.WorkConsumer : work1模式内容为:mq的work模式已经完成6
2023-06-21 09:06:12.059 INFO 21492 --- [ntContainer#4-1] com.mq.mqcloud.controller.WorkConsumer : work1模式内容为:mq的work模式已经完成8解析
- 消费者1和消费者2接收的内容是不同的，同一个消息只能被一个消费者获取
- 消费者1和消费者2获取到的消息数量是相同的，一个是消费奇数号消息，一个是消费偶数号消息
其实，这样是不合理的，因为消费者2线程线程停顿时间短，应该是消费者2要比消费者1获取到的消息更多，这里面涉及到两个知识点，一个是轮询分发，一个是公平分发。
轮询分发:上面的案例就是轮询分发，在默认情况下，RabbitMQ将逐个发送消息到在序列中的下一个消费者(不考虑每个任务的时长等待等，且提前一次性分配，并非一个一个分配)。每个消费者获得相同数量的消息。
公平分发:使用prefetch配置来限制RabbitMQ只发不超过1条的消息给同一个消费者，当消息处理完毕后，有了反馈，才会进行第二次发送
在配置环境中，增加属性prefetch，这个属性的意思是每次领取多少条消息，消费完以后就继续领取
listener:
simple:
acknowledge-mode: auto
#每次领取二个消息，消费以后再领取
prefetch: 2增加属性重新运行可以知道，work2消费了8条，work1消费了一条
2023-06-21 09:48:06.838 INFO 21956 --- [ntContainer#5-1] com.mq.mqcloud.controller.WorkConsumer2 : work2模式内容为:mq的simple模式已经完成3
2023-06-21 09:48:06.838 INFO 21956 --- [ntContainer#5-1] com.mq.mqcloud.controller.WorkConsumer2 : work2模式内容为:mq的simple模式已经完成4
2023-06-21 09:48:06.841 INFO 21956 --- [ntContainer#5-1] com.mq.mqcloud.controller.WorkConsumer2 : work2模式内容为:mq的simple模式已经完成5
2023-06-21 09:48:06.841 INFO 21956 --- [ntContainer#5-1] com.mq.mqcloud.controller.WorkConsumer2 : work2模式内容为:mq的simple模式已经完成7
2023-06-21 09:48:06.841 INFO 21956 --- [ntContainer#5-1] com.mq.mqcloud.controller.WorkConsumer2 : work2模式内容为:mq的simple模式已经完成8
2023-06-21 09:48:06.843 INFO 21956 --- [ntContainer#5-1] com.mq.mqcloud.controller.WorkConsumer2 : work2模式内容为:mq的simple模式已经完成6
2023-06-21 09:48:06.843 INFO 21956 --- [ntContainer#5-1] com.mq.mqcloud.controller.WorkConsumer2 : work2模式内容为:mq的simple模式已经完成9
2023-06-21 09:48:07.828 INFO 21956 --- [ntContainer#4-1] com.mq.mqcloud.controller.WorkConsumer : work1模式内容为:mq的simple模式已经完成0
2023-06-21 09:48:08.830 INFO 21956 --- [ntContainer#4-1] com.mq.mqcloud.controller.WorkConsumer : work1模式内容为:mq的simple模式已经完成1其实也可以通过增加消费端个数属性来实现,按照下面这样配置，通过增加消费端个数，上面的内容都会被work1给消费掉
listener:
simple:
##manual:手动处理 auto:自动处理
acknowledge-mode: auto
#消费端监听个数(即@RabbitListenter开启几个线程去处理)
concurrency: 10
#消费端监听的最大个数
max-concurrency: 10
#每次领取二个消息，消费以后再领取
prefetch: 2
#消费不成功的消息,拒绝入队
default-requeue-rejected: true
retry:
#开启消息重试
enabled: true
#重试次数.
max-attempts: 4
#重试最大间隔时间
max-interval: 10000
#重试初始间隔时间
initial-interval: 2000 3.4 订阅模式(Fanout消息模型)
x:交换机
1、一个生产者，多个消费者
2、每个消费者都有自己的一个队列
3、生产者没有将消息直接发送到队列，而是发送到了交换机
4、每个队列都要绑定到交换机
5、生产者发送的消息，经过交换机，到达队列，实现一个消息被多个消费者获取的目的(一个消费者队列可以有多个消费者实例，只有其中一个消费者实例会被消费)
Producer
@RestController
@AllArgsConstructor
@RequestMapping("/test/mq")
public class FanoutExchangeProducer {
private RabbitTemplate rabbitTemplate;
public static final String EXCHANGE_NAME = "exchange.fanout";
public static final String EXCHANGE_QUEUE_1 = "exchange.fanout.queue_1";
public static final String EXCHANGE_QUEUE_2 = "exchange.fanout.queue_2";
/**
* 扇形交换机，订阅模式
*
* @param context
*/
@GetMapping("/fanout/{context}")
public void sendMessage(@PathVariable String context) {
rabbitTemplate.convertAndSend(FanoutExchangeProducer.EXCHANGE_NAME, "", context);
}
}Consumer
@Component
@Slf4j
public class FanoutExchangeConsumer {
/**
* 订阅模式，扇形交换机
* @param context
*/
@RabbitListener(bindings =
@QueueBinding(exchange = @Exchange(value = FanoutExchangeProducer.EXCHANGE_NAME,type = ExchangeTypes.FANOUT),
value = @Queue(value = FanoutExchangeProducer.EXCHANGE_QUEUE_1)
))
@RabbitHandler
public void exchangeFanoutQueue1(String context){
log.info("通道1接收到的消息为:{}",context);
}
/**
* 订阅模式，扇形交换机
*
* @param context
*/
@RabbitListener(bindings =
@QueueBinding(exchange = @Exchange(value = FanoutExchangeProducer.EXCHANGE_NAME,type = ExchangeTypes.FANOUT) ,
value = @Queue(value = FanoutExchangeProducer.EXCHANGE_QUEUE_2)
))
@RabbitHandler
public void exchangeFanoutQueue2(String context){
log.info("通道2接收到的消息为:{}",context);
}
}结果
消息放到了交换机以后，通道1和通道2都接收到了消息(相当于通道1和通道2都跟这个交换机订阅了消息)
2023-06-21 11:20:43.262 INFO 4696 --- [ntContainer#4-1] c.m.m.controller.FanoutExchangeConsumer : 通道1接收到的消息为:mq的fanout模式(订阅模式)
2023-06-21 11:20:43.262 INFO 4696 --- [ntContainer#3-1] c.m.m.controller.FanoutExchangeConsumer : 通道2接收到的消息为:mq的fanout模式(订阅模式)
3.5 路由模式(Direct消息模型)
路由消息模型是交换机根据routingKey进行消息投递的，每个队列都有自己专属的routingKey，生产者发送消息时，指定交换机和rountingKey,消息到了交换机之后，交换机通过routingKey将消息投递到指定队列
Producer
@RestController
@AllArgsConstructor
@RequestMapping("/test/mq")
public class DirectExchangeProducer {
private RabbitTemplate rabbitTemplate;
public static final String EXCHANGE_DIRECT = "exchange.direct";
public static final String EXCHANGE_DIRECT_QUEUE_1 = "exchange.direct.queue_1";
public static final String EXCHANGE_DIRECT_QUEUE_2 = "exchange.direct.queue_2";
public static final String EXCHANGE_DIRECT_ROUTING_KEY_1 = "exchange.direct.routing.1";
public static final String EXCHANGE_DIRECT_ROUTING_KEY_2 = "exchange.direct.routing.2";
@GetMapping("/direct")
public void sendMessageDirect(@RequestParam("context") String context, @RequestParam("routingkey") Integer routingkey) {
if (1 == routingkey) {
rabbitTemplate.convertAndSend(EXCHANGE_DIRECT, EXCHANGE_DIRECT_ROUTING_KEY_1, context + routingkey);
} else if (2 == routingkey) {
rabbitTemplate.convertAndSend(EXCHANGE_DIRECT, EXCHANGE_DIRECT_ROUTING_KEY_2, context + routingkey);
} else {
System.out.println("数据非法");
}
}
}Consumer
@Component
@Slf4j
public class DirectExchangeConsumer {
@RabbitListener(bindings = @QueueBinding(
exchange = @Exchange(value = DirectExchangeProducer.EXCHANGE_DIRECT, type = ExchangeTypes.DIRECT),
value = @Queue(value = DirectExchangeProducer.EXCHANGE_DIRECT_QUEUE_1),
key = DirectExchangeProducer.EXCHANGE_DIRECT_ROUTING_KEY_1))
public void exchangeDirectRoutingKey1(String context, Message message) {
log.info("key1:" + message.getMessageProperties().getReceivedRoutingKey());
log.info("路由器模式1 接收到的消息为:{}", context);
}
@RabbitListener(bindings = @QueueBinding(
exchange = @Exchange(value = DirectExchangeProducer.EXCHANGE_DIRECT, type = ExchangeTypes.DIRECT),
value = @Queue(value = DirectExchangeProducer.EXCHANGE_DIRECT_QUEUE_2),
key = DirectExchangeProducer.EXCHANGE_DIRECT_ROUTING_KEY_2))
public void exchangeDirectRoutingKey2(String context, Message message) {
log.info("key2:" + message.getMessageProperties().getReceivedRoutingKey());
log.info("路由器模式2 接收到的消息为:{}", context);
}
}结果
2023-06-22 14:48:49.642 INFO 22476 --- [ntContainer#2-2] c.m.m.controller.DirectExchangeConsumer : key2:exchange.direct.routing.2
2023-06-22 14:48:49.643 INFO 22476 --- [ntContainer#2-2] c.m.m.controller.DirectExchangeConsumer : 路由器模式2 接收到的消息为:内容2
2023-06-22 14:48:56.289 INFO 22476 --- [ntContainer#1-1] c.m.m.controller.DirectExchangeConsumer : key1:exchange.direct.routing.1
2023-06-22 14:48:56.289 INFO 22476 --- [ntContainer#1-1] c.m.m.controller.DirectExchangeConsumer : 路由器模式1 接收到的消息为:内容1
2023-06-22 14:48:59.037 INFO 22476 --- [ntContainer#2-2] c.m.m.controller.DirectExchangeConsumer : key2:exchange.direct.routing.2
2023-06-22 14:48:59.038 INFO 22476 --- [ntContainer#2-2] c.m.m.controller.DirectExchangeConsumer : 路由器模式2 接收到的消息为:内容23.6 主题模式(Topic模式，通配符模式)
在Topic模式中，Routingkey不再时固定的字符，而是有了通配符，交换机可以进行模糊匹配队列
RoutingKey一般由一个或者多个单词组成，多个单词以"."分割
"*":匹配一个单词,就只有一个单词 比如topic.* 可以匹配到topic.AB topic.AC
"#":匹配一个或者多个单词 比如topic.# 可以配到到topic.AB也可以匹配到topic.AB.AC
Producer
@RestController
@AllArgsConstructor
@RequestMapping("/test/mq")
public class TopicExchangeProducer {
private RabbitTemplate rabbitTemplate;
public static final String EXCHANGE_TOPIC = "exchange.topic";
public static final String EXCHANGE_TOPIC_QUEUE_1 = "exchange.topic.queue_1";
public static final String EXCHANGE_TOPIC_QUEUE_2 = "exchange.topic.queue_2";
//只要包含了routingkey都会匹配到
public static final String EXCHANGE_TOPIC_ROUTING_KEY_1 = "#.routingkey.#";
//routingkey开头，后面接了一个单词的都会匹配到,比如routing.AB
public static final String EXCHANGE_TOPIC_ROUTING_KEY_2 = "routingkey.*";
public static final String EXCHANGE_TOPIC_CASE_KEY_1 = "topic.routingkey.case1";
//如果case_key_2这样写,那么绑定case_key_1的队列一样会接收到，因为case_key_2也一样和key1匹配上
public static final String EXCHANGE_TOPIC_CASE_KEY_2 = "routingkey.case2";
@GetMapping("/topic")
public void sendMessageDirect(@RequestParam("context") String context, @RequestParam("routingkey") Integer routingkey) {
if (1 == routingkey) {
rabbitTemplate.convertAndSend(EXCHANGE_TOPIC, EXCHANGE_TOPIC_CASE_KEY_1, context + routingkey);
} else if (2 == routingkey) {
rabbitTemplate.convertAndSend(EXCHANGE_TOPIC, EXCHANGE_TOPIC_CASE_KEY_2, context + routingkey);
} else {
System.out.println("数据非法");
}
}
}Consumer
@Component
@Slf4j
public class TopicExchangeConsumer {
@RabbitListener(bindings = @QueueBinding(
exchange = @Exchange(value = TopicExchangeProducer.EXCHANGE_TOPIC, type = ExchangeTypes.TOPIC),
value = @Queue(value = TopicExchangeProducer.EXCHANGE_TOPIC_QUEUE_1),
key = TopicExchangeProducer.EXCHANGE_TOPIC_ROUTING_KEY_1))
@RabbitHandler
public void exchangeTopicRoutingKey1(String context, Message message) {
System.out.println("key1:"+message.getMessageProperties().getReceivedRoutingKey());
log.info("主题模式1:内容为:{}", context);
}
@RabbitListener(bindings = @QueueBinding(
exchange = @Exchange(value = TopicExchangeProducer.EXCHANGE_TOPIC, type = ExchangeTypes.TOPIC),
value = @Queue(value = TopicExchangeProducer.EXCHANGE_TOPIC_QUEUE_2),
key = ExchangeProducer.EXCHANGE_TOPIC_ROUTING_KEY_2))
@RabbitHandler
public void exchangeTopicRoutingKey2(String context, Message message) {
System.out.println("key2:"+message.getMessageProperties().getReceivedRoutingKey());
log.info("主题模式2:内容为:{}", context);
}
}结果1
key1:topic.routingkey.case1
2023-06-22 15:32:10.554 INFO 26060 --- [ntContainer#8-1] c.m.m.controller.TopicExchangeConsumer : 主题模式1:内容为:内容1结果2
因为第二个key也会被第一个匹配到，所以会产生两个结果
key1:routingkey.case2
key2:routingkey.case2
2023-06-22 15:32:25.451 INFO 26060 --- [ntContainer#8-1] c.m.m.controller.TopicExchangeConsumer : 主题模式1:内容为:内容2
2023-06-22 15:32:25.451 INFO 26060 --- [ntContainer#9-1] c.m.m.controller.TopicExchangeConsumer : 主题模式2:内容为:内容2
![[进阶]网络通信：TCP通信-支持与多个客户端同时通信](https://img-blog.csdnimg.cn/2b6c444834844ac1a56f08abeece00a8.png)
