标签：moc 延迟 入队 大数 超级 detach rect 分布 显示
MQ全称为Message Queue,消息队列是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接它们。
消息传递指的是程序之间通过在消息中发送数据进行通信，而不是通过直接调用彼此来通信，直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。
服务与服务之间的通信方式与两种：同步调用和异步消息调用
同步调用：远程过程调用，REST 和 RPC
异步消息调用：消息队列
常见消息队列中间件
RabbitMQ、ActiveMQ、RocketMQ、Kafka
RabbitMQ 稳定可靠,数据一致,支持多协议,有消息确认,基于erlang语言
Kafka 高吞吐,高性能,快速持久化,无消息确认,无消息遗漏,可能会有有重复消息,依赖于zookeeper,成本高.
ActiveMQ 不够灵活轻巧,对队列较多情况支持不好.
RocketMQ 性能好,高吞吐,高可用性,支持大规模分布式，协议支持单一
解耦
场景：用户下单，订单系统需要通知库存系统
传统做法
例:订单系统调用库存系统的接口
传统模式缺点:假如库存系统无法访问，则订单减库存将失败，从而导致订单失败，订单系统与库存系统耦合
使用消息队列
订单系统：用户下单后，订单系统完成持久化处理，将消息写入消息队列，返回用户订单下单成功
库存系统：订阅下单的消息，采用拉/推的方式，获取下单信息，库存系统根据下单信息，进行库存操作
在下单时库存系统不能正常使用。也不影响正常下单，因为下单后，订单系统写入消息队列就不再关心其他的后续操作了。实现订单系统与库存系统的应用解耦
异步
场景:用户注册后，需要发注册邮件和注册短信
传统做法
a.串行：将注册信息写入数据库成功后，发送注册邮件，再发送注册短信
b.并行：将注册信息写入数据库成功后，发送注册邮件的同时，发送注册短信
使用消息队列
将不是必须的业务逻辑，异步处理。
发送注册邮件和发送注册邮件异步读取被写入的消息队列，减少响应时间
流量削峰
场景：商品秒杀业务，一般会因为流量过大，导致流量暴增，应用挂掉
传统做法
限制用户数量
使用消息队列
用户的请求，服务器接收后，首先写入消息队列，秒杀业务根据消息队列中的请求信息，再做后续处理
假如消息队列长度超过最大数量，则直接抛弃用户请求或跳转到错误页面
消息通讯
消息通讯：消息队列一般都内置了高效的通信机制，因此也可以用在纯的消息通讯。比如实现点对点消息队列，或者聊天室等
使用消息队列实现点对点通信
客户端A和客户端B使用同一队列，进行消息通讯
使用消息队列实现聊天室通信
客户端A，客户端B，客户端N订阅同一主题，进行消息发布和接收。实现类似聊天室效果
日志处理
日志处理：指将消息队列用在日志处理中，比如Kafka的应用，解决大量日志传输的问题
使用消息队列完成日志处理
日志采集客户端，负责日志数据采集，定时写受写入Kafka队列
Kafka消息队列，负责日志数据的接收，存储和转发
日志处理应用：订阅并消费kafka队列中的日志数据
RabbitMQ是一个在AMQP基础上完成的，是一个可复用的企业消息系统。遵循Mozilla Public License开源协议。
AMQP，即Advanced Message Queuing Protocol, 一个提供统一消息服务的应用层标准高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。Erlang中的实现有 RabbitMQ等。
特性：
保证可靠性：使用一些机制来保证可靠性，如持久化、传输确认、发布确认
灵活的路由功能
可伸缩性：支持消息集群，多台RabbitMQ服务器可以组成一个集群
高可用性：RabbitMQ集群中的某个节点出现问题时队列任然可用
支持多种协议
支持多语言客户端
提供良好的管理界面
提供跟踪机制：如果消息出现异常，可以通过跟踪机制分析异常原因
提供插件机制：可通过插件进行多方面扩展
使用 VM 基于 Centos 安装
安装前准备
安装 C++ 编译环境
# yum -y install make gcc gcc-c++
yum -y install make gcc gcc-c++ kernel-devel m4 ncurses-devel openssl-devel unixODBC unixODBC-devel httpd python-simplejson
下载 erlang 和 RabbitMQ
# 下载erlang
wget http://www.erlang.org/download/otp_src_20.1.tar.gz
?
# 下载rabbitMQ
wget https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.7.0/rabbitmq-server-generic-unix-3.7.0.tar.xz
安装 erlang
解压 erlang 安装包
tar -xvf otp_src_20.1.tar.gz
进入解压文件夹
cd otp_src_20.1
指定安装目录及安装配置（需要先安装并配置JDK）
?
# erlang指定安装在/usr/local/erlang目录
./configure --prefix=/usr/local/erlang --enable-smp-support --enable-threads --enable-sctp --enable-kernel-poll --enable-hipe --with-ssl --without-javac
?
编译与安装
make && make install
配置erlang环境变量
vi /etc/profile
?
将 export PATH=$PATH:/usr/local/erlang/bin 添加到文件末尾
重新加载profile文件
source /etc/profile
安装 RabbitMQ
解压 RabbitMQ 安装包
由于下载的安装包为xz文件，先将xz解压为tar xz -d rabbitmq-server-generic-unix-3.7.0.tar.xz 再解压缩tar文件 tar -xvf rabbitmq-server-generic-unix-3.7.0.tar
启动 RabbitMQ
进入到解压的 RabbitMQ 的sbin 目录
cd rabbitmq_server-3.7.0/sbin
启动
./rabbitmq-server -detached
查看进程
lsof -i:5672 ps aux|grep rabbit #ps a 显示现行终端机下的所有程序，包括其他用户的程序。 #ps u 以用户为主的格式来显示程序状况。 #ps x 显示所有程序，不以终端机来区分。
启动管理界面
启动 RabbitMQ 的管理系统插件(需进入 sbin 目录)
./rabbitmq-plugins enable rabbitmq_management
访问管理系统
放行端口
如果没有网络指令需要先安装：yum install net-tools
查看并放行端口
netstat -tlnp firewall-cmd --add-port=15672/tcp --permanent firewall-cmd --add-port=5672/tcp --permanent
也可以直接关闭防火墙
CentOS7
#关闭防火墙 systemctl stop firewalld #开机禁用 systemctl disable firewalld #查看状态 systemctl status firewalld
云服务器需要在控制台添加“安全组设置”
用户级别
超级管理员administrator,可以登录控制台,查看所有信息,可以对用户和策略进行操作
监控者monitoring,可以登录控制台,可以查看节点的相关信息,比如进程数,内存磁盘使用情况
策略制定者policymaker ,可以登录控制台,制定策略,但是无法查看节点信息
普通管理员 management 仅能登录控制台
其他, 无法登录控制台,一般指的是提供者和消费者
添加用户
命令模式
添加/配置用户
# 插件目录 ./rabbitmqctl add_user ytao admin123
设置用户权限
#设置admin为administrator级别 ./rabbitmqctl set_user_tags ytao administrator
web方式
浏览器访问 RabbitMQ web界面 http:// IP地址:15672/(使用guest guest 登录, guest 具有最高权限)
添加用户
Admin --> Add a user 填写用户信息 --> Add user 提交
为用户分配可以访问的虚拟主机
默认情况下没有任何可以访问的,我们可以添加一个主机(相当于添加一个数据库),然后分配权限
创建虚拟主机
点击菜单(Virtual Hosts) --> Add a new Virtual host(添加新的虚拟主机) --> Name(主机名) --> 点击创建(Add virtual host)
给指定用户分配虚拟主机
点击Users --> 点击用户名 --> 选择虚拟主机 --> 点击设置权限(Set permission)
设置完成后，回到用户界面确认
简单模式
简单模式就是我们的生产者将消息发到队列,消费者从队列中取消息,一条消息对应一个消费者
工作模式
Work模式就是一条消息可以被多个消费者尝试接收,但是最终只能有一个消费者能获取
订阅模式 一条消息可以被多个消费者同时获取,生产者将消息发送到交换机,消费者将自己对应的队列注册到交换机,当发送消息后所有注册的队列的消费者都可以收到消息
路由模式
生产者将消息发送到了type为direct模式的交换机，消费者的队列在将自己绑定到路由的时候会给自己绑定一个key，只有消费者发送对应key格式的消息时候队列才会收到消息
Topic 模式
RPC 模式
创建 Maven 项目
添加依赖
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>4.5.0</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.7.25</version>
</dependency>
创建日志配置文件 log4j.properties
log4j.rootLogger=DEBUG,A1 log4j.logger.com.taotao = DEBUG
log4j.logger.org.mybatis = DEBUG
log4j.appender.A1=org.apache.log4j.ConsoleAppender
log4j.appender.A1.layout=org.apache.log4j.PatternLayout
log4j.appender.A1.layout.ConversionPattern=%-d{yyyy-MM-dd HH:mm:ss,SSS} [%t] [%c]-[%p] %m%n
创建帮助类
package com.qfedu.utils;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class ConnectionUtil {
public static Connection getConnection() throws Exception {
//定义连接工厂
ConnectionFactory factory = new ConnectionFactory();
// 设置服务地址
factory.setHost("服务器地址");
//端口
factory.setPort(5672);
// 设置账号信息，用户名、密码、vhost
factory.setVirtualHost("账号信息");
factory.setUsername("用户名");
factory.setPassword("密码");
// 通过工程获取连接
Connection connection = factory.newConnection();
return connection;
}
}
发送消息
import com.qfedu.utils.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
public class Send {
public static void main(String[] argv) throws Exception {
// 获取到连接以及mq通道
Connection connection = ConnectionUtil.getConnection();
// 相当于数据库中的创建连接
// 从连接中创建通道
Channel channel = connection.createChannel();
// 相当于数据库中的 statement
// 声明（创建）队列,如果存在就不创建,不存在就创建
// 参数1 队列名,
// 参数2 durable： 是否持久化, 队列的声明默认是存放到内存中的，如果rabbitmq重启会丢失，如果想重启之后还存在就要使队列持久化，保存到Erlang自带的Mnesia数据库中，当rabbitmq重启之后会读取该数据库
// 参数3 exclusive：是否排外的，有两个作用，一：当连接关闭时connection.close()该队列是否会自动删除； 二：该队列是否是私有的private，如果不是排外的，可以使用两个消费者都访问同一个队列，没有任何问题，如果是排外 的，会对当前队列加锁，其他通道channel是不能访问的，如果强制访问会报异常： com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=405, reply-text=RESOURCE_LOCKED - cannot obtain exclusive access to locked queue ‘queue_name‘ in vhost ‘/‘, class-id=50, method-id=20)一般等于true的话 用于一个队列只能有一个消费者来消费的场景
// 参数4 autoDelete：是否自动删除，当最后一个消费者断开连接之后队列是否自动被删除，可以通过RabbitMQ Management，查看某个队列的消费者数量，当consumers = 0时队列就会自动删除
// 参数5 arguments： 参数
//channel.queueDeclare("queue1", false, false, true, null);
// 消息内容
String message = "Hello World!";
// 参数1 交换机,此处无
// 参数2 发送到哪个队列
// 参数3 属性
// 参数4 内容
channel.basicPublish("", "queue1", null, message.getBytes());
// 将消息发动到数据库
System.out.println(" 发送数据 ‘" + message + "‘");
//关闭通道和连接
channel.close();
connection.close();
}
}
消费消息
import com.qfedu.utils.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
public class Send {
public static void main(String[] argv) throws Exception {
// 获取到连接以及mq通道
Connection connection = ConnectionUtil.getConnection();
// 相当于数据库中的创建连接
// 从连接中创建通道
Channel channel = connection.createChannel();
// 相当于数据库中的 statement
// 声明（创建）队列,如果存在就不创建,不存在就创建
// 参数1 队列名,
// 参数2 durable： 是否持久化, 队列的声明默认是存放到内存中的，如果rabbitmq重启会丢失，如果想重启之后还存在就要使队列持久化，保存到Erlang自带的Mnesia数据库中，当rabbitmq重启之后会读取该数据库
// 参数3 exclusive：是否排外的，有两个作用，一：当连接关闭时connection.close()该队列是否会自动删除； 二：该队列是否是私有的private，如果不是排外的，可以使用两个消费者都访问同一个队列，没有任何问题，如果是排外 的，会对当前队列加锁，其他通道channel是不能访问的，如果强制访问会报异常： com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=405, reply-text=RESOURCE_LOCKED - cannot obtain exclusive access to locked queue ‘queue_name‘ in vhost ‘/‘, class-id=50, method-id=20)一般等于true的话 用于一个队列只能有一个消费者来消费的场景
// 参数4 autoDelete：是否自动删除，当最后一个消费者断开连接之后队列是否自动被删除，可以通过RabbitMQ Management，查看某个队列的消费者数量，当consumers = 0时队列就会自动删除
// 参数5 arguments： 参数
//channel.queueDeclare("queue1", false, false, true, null);
// 消息内容
String message = "Hello World!";
// 参数1 交换机,此处无
// 参数2 发送到哪个队列
// 参数3 属性
// 参数4 内容
channel.basicPublish("", "queue1", null, message.getBytes());
// 将消息发动到数据库
System.out.println(" 发送数据 ‘" + message + "‘");
//关闭通道和连接
channel.close();
connection.close();
}
}
消息发送者
创建 SpringBoot 项目，导入以下依赖：
Spring Web
Thymeleaf
Spring for RabbitMQ
在项目中新增RabbitMQ支持导入以下依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit-test</artifactId>
<scope>test</scope>
</dependency>
在application.yml配置RabbitMQ服务器连接属性
spring:
application:
name: mq-sender-demo
rabbitmq:
host: 服务器地址
port: 5672
username: 用户名
password: 密码
virtual-host: 用户信息
server:
port: 9001
配置RabbitMQ创建队列（Quence）
@Configuration
public class RabbitMQConfig {
@Bean
public Queue queue9() {
return new Queue("queue9");
}
@Bean
public Queue queue10() {
return new Queue("queue10");
}
/**
* 声明交换机,fanout 类型
*/
@Bean
public FanoutExchange fanoutExchange() {
FanoutExchange fanoutExchange = new FanoutExchange("fanoutExchange");
return fanoutExchange;
}
/**
* 将队列和交换机绑定
*/
@Bean
public Binding bindingFanoutExchange1(Queue queue9, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(queue9).to(fanoutExchange);
}
@Bean
public Binding bindingFanoutExchange2(Queue queue10, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(queue10).to(fanoutExchange);
}
@Bean
public Queue queue11() {
return new Queue("queue11");
}
@Bean
public Queue queue12() {
return new Queue("queue12");
}
/**
* 声明交换机,direct 类型
*/
@Bean
public DirectExchange directExchange() {
DirectExchange directExchange = new DirectExchange("directExchange");
return directExchange;
}
/**
* 将队列和交换机绑定
*/
@Bean
public Binding bindingDirectExchange(Queue queue11, DirectExchange directExchange) {
return BindingBuilder.bind(queue11).to(directExchange).with("rk1");
}
@Bean
public Binding bindingDirectExchange2(Queue queue12, DirectExchange directExchange) {
return BindingBuilder.bind(queue12).to(directExchange).with("rk2");
}
}
在消息发送者中注入AmqpTemplate对象即可发送消息
@Service
public class SendMsgService {
@Autowired
private AmqpTemplate amqpTemplate;
public void sendMsg(String message){
amqpTemplate.convertAndSend("wfx-simple",message);
//amqpTemplate.convertAndSend("fanout", "", message);
//amqpTemplate.convertAndSend("direct", "rk1", message);
}
}
消息接收者
监听队列
@Component
@RabbitListener(queues = "wfx-direct-quence1")
public class ReceiveDirectHanlder1 {
@RabbitHandler//标记当前方法是用来处理消息的
public void recevieMsg(String message) {
System.out.println("收到wfx-direct-quence1消息: =>"+message);
}
}
将对象转换成字符串进行传递
对象序列号实现
bean 实现序列号接口
生产者消费者的 bean 包名必须一致
消息生产者
@Service
public class SendMsgService {
@Autowired
private AmqpTemplate amqpTemplate;
public void sendMsg(Goods goods) {
byte[] bytes = SerializationUtils.serialize(goods);
amqpTemplate.convertAndSend("queue1",bytes);
}
}
消息消费者
@Service
@RabbitListener(queues = "queue1")
public class ReviceMsgService {
@RabbitHandler
public void receiveMsg(byte[] bs) {
Goods goods = (Goods) SerializationUtils.deserialize(bs);
System.out.println(goods);
}
}
JSON 字符串实现
消费生产者
@Service
public class SendMsgService {
@Autowired
private AmqpTemplate amqpTemplate;
public void sendMsg(Goods goods) throws JsonProcessingException {
ObjectMapper mapper=new ObjectMapper();
String message=mapper.writeValueAsString(goods);
amqpTemplate.convertAndSend("queue1",message);
}
}
消息消费者
@Service
@RabbitListener(queues = "queue1")
public class ReviceMsgService {
@RabbitHandler
public void receiveMsg(String msg) throws JsonProcessingException {
ObjectMapper mapper=new ObjectMapper();
Goods goods=mapper.readValue(msg,Goods.class);
System.out.println(goods);
}
}
Java 代码管理
交换机、队列创建及绑定
//1. 声明队列-HelloWorld
//参数1：queue - 指定队列的名称
//参数2：durable - 当前队列是否需要持久化（true）
//参数3：exclusive - 是否排外（conn.close() - 当前队列会被自动删除，当前队列只能被一个消费者消费）
//参数4：autoDelete - 如果这个队列没有消费者在消费，队列自动删除
//参数5：arguments - 指定当前队列的其他信息
channel.queueDeclare("HelloWorld",true,false,false,null);
//2. 创建exchange - 绑定某一个队列
//参数1： exchange的名称
//参数2： 指定exchange的类型 FANOUT - pubsub , DIRECT - Routing , TOPIC - Topics
channel.exchangeDeclare("pubsub-exchange", BuiltinExchangeType.FANOUT);
channel.queueBind("pubsub-queue1","pubsub-exchange","");
channel.queueBind("pubsub-queue2","pubsub-exchange","");
SpringBoot Java 配置管理
配置RabbitMQ创建队列（Quence）
@Configuration
public class RabbitMQConfiguration {
@Bean
public Queue queue() {
return new Queue("wfx-quence");
}
@Bean
public Queue fanoutQuence() {
return new Queue("wfx-fanout-quence");
}
/**
* 声明交换机,fanout 类型
*/
@Bean
public FanoutExchange fanoutExchange() {
FanoutExchange fanoutExchange = new FanoutExchange("fanoutExchange");
return fanoutExchange;
}
/**
* 将队列和交换机绑定
*/
@Bean
public Binding bindingFanoutExchange(Queue fanoutQuence, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(fanoutQuence).to(fanoutExchange);
}
@Bean
public Queue directQuence1() {
return new Queue("wfx-direct-quence1");
}
@Bean
public Queue directQuence2() {
return new Queue("wfx-direct-quence2");
}
/**
* 声明交换机,direct 类型
*/
@Bean
public DirectExchange directExchange() {
DirectExchange directExchange = new DirectExchange("directExchange");
return directExchange;
}
/**
* 将队列和交换机绑定
*/
@Bean
public Binding bindingDirectExchange(Queue directQuence1, DirectExchange directExchange) {
return BindingBuilder.bind(directQuence1).to(directExchange).with("rk1");
}
@Bean
public Binding bindingDirectExchange2(Queue directQuence2, DirectExchange directExchange) {
return BindingBuilder.bind(directQuence2).to(directExchange).with("rk2");
}
}
RabbitMQ 事务
//1.开启事务 channel.txSelect(); //2.提交事务 channel.txCommit(); //3.事务回滚 channel.txRollback();
消息确认机制
消息确认机制和 return 机制
Maven 项目的消息确认
普通 confirm 方式
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
String message = "Hello World!";
//1.开启消息确认
channel.confirmSelect();
//2.发送消息
channel.basicPublish("ex2", "c", null, message.getBytes());
//3.获取确认
boolean b = channel.waitForConfirms();
System.out.println("消息发送"+(b?"成功":"失败"));
channel.close();
connection.close();
批量 confirm 方式
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
String message = "Hello World!";
//1.开启消息确认
channel.confirmSelect();
//2.发送消息
for (int i=1; i<=10; i++) {
message += i;
channel.basicPublish("ex2", "c", null, message.getBytes());
}
//3.批量确认:发送的所有消息中有一个失败就直接全部失败，抛出IO异常
boolean b = channel.waitForConfirms();
channel.close();
connection.close();
异步 confirm 方式
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
String message = "Hello World!";
//1.开启消息确认
channel.confirmSelect();
//2.发送消息
for (int i=1; i<=10; i++) {
message += i;
channel.basicPublish("ex2", "c", null, message.getBytes());
}
//3.开启异步confirm
channel.addConfirmListener(new ConfirmListener() {
//参数l表示返回的消息标识，参数b表示是否为批量confirm
public void handleAck(long l, boolean b) throws IOException {
System.out.println("----消息发送成功");
}
public void handleNack(long l, boolean b) throws IOException {
System.out.println("----消息发送失败");
}
});
channel.close();
connection.close();
return 机制
发送消息之前开启return机制
发送消息时指定mandatory参数为true
由于return机制是异步处理，所以在发送消息之后不关闭channel
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
String message = "Hello World!";
channel.addReturnListener(new ReturnListener() {
public void handleReturn(int i, String s, String s1, String s2, AMQP.BasicProperties basicProperties,
byte[] bytes) throws IOException {
System.out.println("消息未分发到队列中");
}
});
//channel.basicPublish("", "queue1", null, message.getBytes());
channel.basicPublish("ex2", "c", true,null, message.getBytes());
springboot 应用消息确认
配置 application.yml
spring:
rabbitmq:
publisher-confirm-type: simple
publisher-returns: true
开启 confirm 和 return 监听
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
@Component
public class PublisherConfireAndReturnConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
Logger logger = LoggerFactory.getLogger(PublisherConfireAndReturnConfig.class);
@Resource
private RabbitTemplate rabbitTemplate;
@PostConstruct
public void initMethod(){
rabbitTemplate.setConfirmCallback(this);
rabbitTemplate.setReturnCallback(this);
}
@Override
public void confirm(CorrelationData correlationData, boolean ack, String s) {
if(ack){
logger.info("--------消息发送(到交换机)成功");
}else{
logger.warn("--------消息发送(到交换机)失败");
}
}
@Override
public void returnedMessage(Message message, int i, String s, String s1, String s2) {
logger.info("~~~~~~~~消息发送到交换机但未分发到队列！！！");
}
}
避免消息重复消费
重复消费消息，会对非幂等行操作造成问题 重复消费消息的原因是，消费者没有给RabbitMQ一个ack
为了解决消息重复消费的问题，可以采用Redis，在消费者消费消息之前，现将消息的id放到Redis中，
id-0（正在执行业务）
id-1（执行业务成功）
如果ack失败，在RabbitMQ将消息交给其他的消费者时，先执行setnx，如果key已经存在，获取他的值，如果是0，当前消费者就什么都不做，如果是1，直接ack。
极端情况：第一个消费者在执行业务时，出现了死锁，在setnx的基础上，再给key设置一个生存时间。
Maven 项目避免重复消费
生产者，发送消息时，指定 messageld
AMQP.BasicProperties properties = new AMQP.BasicProperties().builder()
.deliveryMode(1) //指定消息书否需要持久化 1 - 需要持久化 2 - 不需要持久化
.messageId(UUID.randomUUID().toString())
.build();
String msg = "Hello-World！";
channel.basicPublish("","HelloWorld",true,properties,msg.getBytes());
消费者，在消费消息时，根据具体业务逻辑去操作 redis
DefaultConsumer consume = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
Jedis jedis = new Jedis("192.168.199.109",6379);
String messageId = properties.getMessageId();
//1. setnx到Redis中，默认指定value-0
String result = jedis.set(messageId, "0", "NX", "EX", 10);
if(result != null && result.equalsIgnoreCase("OK")) {
System.out.println("接收到消息：" + new String(body, "UTF-8"));
//2. 消费成功，set messageId 1
jedis.set(messageId,"1");
channel.basicAck(envelope.getDeliveryTag(),false);
}else {
//3. 如果1中的setnx失败，获取key对应的value，如果是0，return，如果是1
String s = jedis.get(messageId);
if("1".equalsIgnoreCase(s)){
channel.basicAck(envelope.getDeliveryTag(),false);
}
}
}
};
springboot 应用避免重复消费
导入依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
编写配置文件
spring:
redis:
host: 47.96.11.185
port: 6379
修改生产者
@Test
void contextLoads() throws IOException {
CorrelationData messageId = new CorrelationData(UUID.randomUUID().toString());
rabbitTemplate.convertAndSend("boot-topic-exchange","slow.red.dog","红色大狼狗！！",messageId);
System.in.read();
}
修改消费者
@Autowired
private StringRedisTemplate redisTemplate;
?
?
@RabbitListener(queues = "boot-queue")
public void getMessage(String msg, Channel channel, Message message) throws IOException {
//0. 获取MessageId
String messageId = message.getMessageProperties().getHeader("spring_returned_message_correlation");
//1. 设置key到Redis
if(redisTemplate.opsForValue().setIfAbsent(messageId,"0",10, TimeUnit.SECONDS)) {
//2. 消费消息
System.out.println("接收到消息：" + msg);
?
//3. 设置key的value为1
redisTemplate.opsForValue().set(messageId,"1",10,TimeUnit.SECONDS);
//4. 手动ack
channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
}else {
//5. 获取Redis中的value即可 如果是1，手动ack
if("1".equalsIgnoreCase(redisTemplate.opsForValue().get(messageId))){
channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
}
}
}
RabbitMQ 延迟机制(TTL) -- SpringBoot
延迟队列概念
什么是延迟队列
延迟队列存储的对象肯定是对应的延时消息，所谓”延时消息”是指当消息被发送以后，并不想让消费者立即拿到消息，而是等待指定时间后，消费者才拿到这个消息进行消费。
RabbitMQ 如何实现延迟队列？
AMQP协议和RabbitMQ队列本身没有直接支持延迟队列功能，但是可以通过TTL（Time To Live）特性模拟出延迟队列的功能。
消息的TTL
消息的TTL就是消息的存活时间。RabbitMQ可以对队列和消息分别设置TTL。对队列设置就是队列没有消费者连着的保留时间，也可以对每一个单独的消息做单独的设置。超过了这个时间，我们认为这个消息就死了，称之为死信。可以通过设置消息的expiration字段或者x-message-ttl属性来设置时间.
实现延迟队列
延迟任务通过消息的TTL来实现。我们需要建立2个队列，一个用于发送消息，一个用于消息过期后的转发目标队列。
生产者输出消息到Queue1，并且这个消息是设置有有效时间的，比如60s。消息会在Queue1中等待60s，如果没有消费者收掉的话，它就是被转发到Queue2，Queue2有消费者，收到处理延迟任务。
创建延迟交换机
创建路由交换机
创建死信队列
创建死信转发队列
交换机队列绑定
SpringBoot 实现延迟队列
添加 MQ 依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit-test</artifactId>
<scope>test</scope>
</dependency>
在 application.yaml 配置 RabbitMQ 服务器连接属性
spring:
application:
name: mq-sender-demo
rabbitmq:
host: 47.96.11.185
port: 5672
username: ytao
password: admin123
virtual-host: wfx_host
# 手动ACK 不开启自动ACK模式,目的是防止报错后未正确处理消息丢失 默认 为 none
listener:
simple:
acknowledge-mode: manual
@Component
@Slf4j
public class RabbitProduct{
?
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendDelayMessage(List<Integer> list) {
//这里的消息可以是任意对象，无需额外配置，直接传即可
log.info("===============延时队列生产消息====================");
log.info("发送时间:{},发送内容:{}", LocalDateTime.now(), list.toString());
this.rabbitTemplate.convertAndSend(
"delay_exchange",
"delay_key",
list,
message -> {
//注意这里时间要是字符串形式
message.getMessageProperties().setExpiration("60000");
return message;
}
);
log.info("{}ms后执行", 60000);
}
}
创建消费者
@Component
@Slf4j
public class RabbitConsumer {
@Autowired
private CcqCustomerCfgService ccqCustomerCfgService;
?
/**
* 默认情况下,如果没有配置手动ACK, 那么Spring Data AMQP 会在消息消费完毕后自动帮我们去ACK
* 存在问题：如果报错了,消息不会丢失,但是会无限循环消费,一直报错,如果开启了错误日志很容易就吧磁盘空间耗完
* 解决方案：手动ACK,或者try-catch 然后在 catch 里面将错误的消息转移到其它的系列中去
* spring.rabbitmq.listener.simple.acknowledge-mode = manual
* @param list 监听的内容
*/
@RabbitListener(queues = "receive_queue")
public void cfgUserReceiveDealy(List<Integer> list, Message message, Channel channel) throws IOException {
log.info("===============接收队列接收消息====================");
log.info("接收时间:{},接受内容:{}", LocalDateTime.now(), list.toString());
//通知 MQ 消息已被接收,可以ACK(从队列中删除)了
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
try {
dosomething.....
} catch (Exception e) {
log.error("============消费失败,尝试消息补发再次消费!==============");
log.error(e.getMessage());
/**
* basicRecover方法是进行补发操作，
* 其中的参数如果为true是把消息退回到queue但是有可能被其它的consumer(集群)接收到，
* 设置为false是只补发给当前的consumer
*/
channel.basicRecover(false);
}
}
}
标签：moc 延迟 入队 大数 超级 detach rect 分布 显示
原文地址：https://www.cnblogs.com/zhu-zhu/p/14333292.html
