角色
生产者
消息的创建者,负责创建和推送数据到消息服务器
消费者
消息的接收方,用于处理数据和确认消息
组件
交换机(exchange) 接受消息 ,分配消息 ,用于存储生产者的消息
路由键(RountingKey) 用于那生成这的数据分配到固定的交换机上面 BingKey把交换机的消息绑定到队列上
队列(Queue)与交换机相连,交换机传递消息的载体
绑定键(BingKey)交换机与队列的绑定Key
信道(Channel)推送消息使用的通道
队列模式
direct 单一 一对一
fanout 广播 一对多
topic 主体 可以一对一,也可以一对多
操作
接受消息的类上面添加注解 @RabbitListener(queues = "QUEUE_B")
rabbitTemplate.convertAndSend(exchange,routeKey,list'); 指定交换机名称,绑定键,数据
@RabbitListener(bindings = @QueueBinding( value = @Queue(value = "QUEUE_E", durable = "false", autoDelete = "true"),exchange = @Exchange(value = "EXCHANGE_E", type = ExchangeTypes.TOPIC),key = "EXCHANGE_E_TO_E")) 指定交换机类型,队列名称,绑定键
消息避免丢失
1、消息进行持久化存储
2、开启Ack确认模式
实施
/**
*@description : 在 ackMode = "MANUAL" 后面添加 concurrency ="1" 表示这个消息固定为4个消费者
*@author : wangyuhua TODO 正常的业务逻辑里面的date是方法执行后的返回结果,data是消费者方法执行后的返回值,根据返回值判断
* todo 根据返回值判断是需要重新进入队列还是直接放弃
*@date : 2022/6/24 14:39
*@param : [data, deliveryTag, channel]
*@return : void
**/
@RabbitListener(bindings = @QueueBinding(value = @Queue(value = "QUEUE_F", durable = "false", autoDelete = "true"),
exchange = @Exchange(value = "EXCHANGE_F", type = ExchangeTypes.TOPIC), key = "EXCHANGE_TO_F"), ackMode = "MANUAL")
public void consumerDoAck(List data, @Header( AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel)
throws IOException {
System.out.println("消息传递到了这里,在这里模拟消费失败,进行重新消费,测试ack");
if (data.contains("999")) {
i+=1;
// RabbitMQ的ack机制中,第二个参数返回true,表示需要将这条消息投递给其他的消费者重新消费
// channel.basicAck(deliveryTag, false);
channel.basicAck(deliveryTag,false);
System.out.println(" 走到了第一个方法的选择里面 " );
} else if (i<3){
i+=1;
System.out.println("开始第"+i+"次消费" );
// 第三个参数true,表示这个消息会重新进入队列
channel.basicNack(deliveryTag, false, true);
}
else {
System.out.println("尝试次数超过三回,放弃该请求");
channel.basicNack(deliveryTag, false, false);
}
}
3、设置集群
4、消息补偿机制
优点
1、异步处理 并行方式,提高吞吐量
2、解耦
3、可以通过规定队列的长度,控制请求量,缓解高并发
4、日志
5、特殊的消息机制,点对点,广播模式等
解决消息消费顺序
创建一个一对一的队列即可
工作模式
三.RabbitMQ的六种工作模式
1.简单模式:只有一个队列一个消费者
简单模式下我们无需指定交换机,RabbitMQ会通过默认的default AMQP交换机将我们的消息投递到指定的队列,它是一种Direct类型的交换机,队列与它绑定时的binding key其实就是队列的名称
2.WorkQueues:工作队列模式,只有一个队列,有多个消费者
工作队列模式也是采用默认的default AMQP交换机,Queue中的消息会被平均分发给多个消费者处理。有两种消息分发方式:
轮询分发:一个消费者消费一条,按均分配,woek模式下默认是采用轮询分发方式。轮询分发就不写代码演示了,比较简单,比如生产费者发送了6条消息到队列中,如果有3个消费者同时监听着这一个队列,那么这3个消费者每人就会分得2条消息。下面主要介绍公平分发。
公平分发:根据消费者的消费能力进行公平分发,处理得快的分得多,处理的慢的分得少,能者多劳。
3.Publish/Subscribe:发布订阅模式,有N个队列,N个消费者,需要声明绑定交换机
RabbitMQ中交换机有三种:fanout、direct、topic
(1)fanout:扇出模式,消息广播到所有于交换机绑定的队列当中
(2)direct:路由模式,消息发布者指定一个key,消息路由到所有匹配key的队列当中
(3)topic:主题模式,与路由模式差不多,可以通过通配符进行模糊匹配
4.RPC模式:支持生产者和消费者不在同一个系统中,即允许远程调用的情况
RPC模式下通常消费者作为服务端,放置在远程的系统中,提供接口;生产者调用接口,并发送消息。
RPC模式是一种远程调用的模式,因为需要http请求,因此速度比系统内部调用慢。而且rpc模式下,通常不易区分哪些是来自外部的请求,哪些是内部的请求,导致整体速度较慢。因此,不能滥用rpc模式。
在MessageProperties中有两个属性:
reply_to:用于定义回调队列的名字
correlation_id:用于关联rpc的消息请求发送与消息响应接收。
要实现RPC模式,生产者需要发送回调队列,工作流程:
1、生产者(Client)开始生产消息后,创建了匿名的、独占的回调队列。
2、生产者(Client)发送请求时,包含两个属性:reply_to,即回调队列;correlation_id,即本次请求的ID
3、请求(request )被发送到rpc_queue队列。
4、消费者(The RPC worker)在rpc_queue上等待请求,收到时,它处理消息并使用replyTo字段中的队列将结果发回客户机
5、生产者在回调队列上等待消息,当消息出现时,校验correlation_id,如果匹配请求中的值则向程序返回该响应数据。
标签:false,消费者,队列,RabbitMq,交换机,消息,模式 From: https://www.cnblogs.com/wyhqmm/p/17443612.html