MQ(message queue) 本质是一个队列，先进先出
1.流量削峰
流量高峰期，将请求订单分多次进行处理，防止服务器崩溃,但是会影响一定的用户体验
2.应用解耦
在多模块系统中，一个应用有多个模块系统，当一个系统出现故障时，就会造成操作异常，
使用消息队列来等待系统修复，不会将故障传送给客户
3.异步处理
两个系统可以使用消息队列进行监听消息，不需要进行等待，让mq进行监听即可,可以自己做自己的事
通俗的讲:
RabbitMQ就像是一个快递站，进行消息的接收，存储和转发
交换机就像是快递员，队列就像是发快递的路线
1.生产者
2.交换机
3.队列
4.消费者
因为RabbitMQ是erlang语言编写的，所以需要erlang环境
https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.8.16/rabbitmq-server-3.8.8-1.el7.noarch.rpm
rpm -ivh erlang-21.3-1.el7.x86_64.rpm
yum install socat -y
rpm -ivh rabbitmq-server-3.8.8-1.el7.noarch.rpm
# 添加开机自启
chkconfig rabbitmq-server on
# 启动服务
/sbin/service rabbitmq-server start
# 查看服务状态
/sbin/service rabbitmq-server status
# 停止服务
/sbin/service rabbitmq-server stop
# 创建新用户
# 账号
rabbitmqctl add_user admin 123
# 设置用户角色
rabbitmqctl set_user_tags admin administrator
# 设置用户权限
set_perrmissions [-p <vhostpath>] <user> <conf> <write> <read>
rabbitmqctl set_perrmissions -p "/" admin ".*" ".*" ".*"
读写权限
# 查看用户和角色
rabbitmqctl list_users
# 关闭防火墙
systemctl stop firewalld
# 访问
http://ip:15672/#/
JMS即Java消息服务(JavaMessage Service)应用程序接口，是一个Java平台中关于面向消息中间件的API
javaEE的一种规范，类似于jdbc
<!--rabbitmq依赖客户端-->
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.8.0</version>
</dependency>
<!--操作文件流-->
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
<version>2.4</version>
</dependency>
连接工厂 >> 设置参数 >> 工厂创建连接 >> 连接中创建信道 >> 信道中声明队列 >> 发送消息
package com.rabbit.review.one;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.nio.charset.StandardCharsets;
/**
* @author zwj
* @date 2022/2/12 - 14:47
* 生产者 复习
*/
public class Producer {
//队列名称
public static final String QUEUE_NAME = "hello";
public static void main(String[] args) throws Exception {
//创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
//设置创建连接的ip地址
factory.setHost("****");
//默认端口5672
//设置用户名密码
factory.setUsername("admin");
factory.setPassword("***");
//获取一个连接
Connection connection = factory.newConnection();
//获取信道
Channel channel = connection.createChannel();
//生成一个队列
/**
* 参数
* 1.队列名称
* 2.是否持久化
* 3.是否可以多个消费者消费
* 4.最后一个消费者断开连接后，是否自动删除
* */
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
//发消息
//生产者通过信道发送消息，声明使用哪个交换机和队列,因为此次没有交换机，所以直接使用默认交换机
String message = "hello world";
channel.basicPublish("",QUEUE_NAME,null,message.getBytes(StandardCharsets.UTF_8));
connection.close();
}
}
package com.rabbit.review.one;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;
/**
* @author zwj
* @date 2022/2/12 - 15:02
* 消费者 初学
*/
public class Consumer {
private static final String QUEUE_NAME = "hello";
public static void main(String[] args) throws IOException, TimeoutException {
//创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
//设置连接ip
factory.setHost("192.168.229.128");
//设置账户密码
factory.setUsername("admin");
factory.setPassword("admin");
//创建一个连接
Connection connection = factory.newConnection();
//创建一个信道
Channel channel = connection.createChannel();
//在信道中声明队列
// channel.queueDeclare(QUEUE_NAME,false,false,false,null);
//消费者接收消息
/**
* 参数
* 1.队列名
* 2.是否自动应答
* 3.成功的回调
* 4.取消消费的回调
* */
DeliverCallback deliverCallback = (consumerTag, message) -> {
System.out.println(new String(message.getBody()));
};
CancelCallback cancelCallback = (consumerTag) -> {
System.out.println("消费中断");
};
channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
// connection.close();
}
}
多个工作线程(消费者)同时工作，消息只会被消费一次，默认轮询接收消息
创建RabbitMQ连接工具类
package com.rabbit.review.util;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/**
* @author zwj
* @date 2022/2/12 - 15:32
* 连接创建信道工具类
*/
public class RabbitMqUtil {
public static Channel getChannel() throws Exception{
//创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
//设置参数
factory.setHost("192.168.229.128");
factory.setUsername("admin");
factory.setPassword("admin");
//创建一个连接
Connection connection = factory.newConnection();
//创建一个信道返回
return connection.createChannel();
}
}
工作线程代码
package com.rabbit.review.workqueue;
import com.rabbit.review.util.RabbitMqUtil;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.Delivery;
import java.nio.charset.StandardCharsets;
/**
* @author zwj
* @date 2022/2/12 - 15:37
*/
public class worker001 {
//队列名称
public static final String QUEUE_NAME = "hello";
public static void main(String[] args) throws Exception{
//工具类创建信道
Channel channel = RabbitMqUtil.getChannel();
//消息接收
DeliverCallback deliverCallback = (consumerTag, message) -> {
System.out.println(new String(message.getBody(), StandardCharsets.UTF_8));
};
//消息取消消费
CancelCallback cancelCallback = (consumerTag) -> {
System.out.println(consumerTag+"消息被取消消费");
};
//消息应答队列
channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
}
}
开启多个线程，使用idea可以讲一个类运行多个线程
工作线程生产者代码
package com.rabbit.review.workqueue;
import com.rabbit.review.util.RabbitMqUtil;
import com.rabbitmq.client.Channel;
import java.nio.charset.StandardCharsets;
/**
* @author zwj
* @date 2022/2/12 - 15:51
* 工作线程 生产者
*/
public class Task001 {
//队列名称
public static final String QUEUE_NAME = "hello";
public static void main(String[] args) throws Exception {
//创建队列
Channel channel = RabbitMqUtil.getChannel();
//声明队列
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
//发送消息
String message = "work queue";
channel.basicPublish("",QUEUE_NAME,null,message.getBytes(StandardCharsets.UTF_8));
System.out.println("消息发送完毕");
channel.getConnection().close();
}
}
生产者发送的消息从不会直接到达队列，都会经过交换机
交换机必须确切直到如何处理收到的消息，由交换机的类型决定
默认交换机 使用空串表示
分类
1.direct(直接)(路由模式)
2.topic(主题)(主题模式)
3.headers(标头)基本已经不用了
4.fanout(扇出)(发布订阅模式)
临时队列
String queueName = channel.queueDeclare().getQueue()
绑定binding
交换机和队列之间的绑定关系
类似于广播
生产者
package com.rabbit.review.fanout;
import com.rabbit.review.util.RabbitMqUtil;
import com.rabbitmq.client.Channel;
import java.nio.charset.StandardCharsets;
/**
* @author zwj
* @date 2022/2/13 - 13:52
* fanout 生产者
*/
public class SendLogs {
public static final String EXCHANGE_NAME = "logs";
public static void main(String[] args) throws Exception {
//创建信道
Channel channel = RabbitMqUtil.getChannel();
//声明交换机
channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
//发送消息
String message = "fanout exchange";
/**
* 参数
* 1.交换机名称
* 2.routingKey 交换机和队列之间的binding
* */
channel.basicPublish(EXCHANGE_NAME,"",null,message.getBytes(StandardCharsets.UTF_8));
System.out.println("消息发送成功");
}
}
消费者
package com.rabbit.review.fanout;
import com.rabbit.review.util.RabbitMqUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.Delivery;
/**
* @author zwj
* @date 2022/2/13 - 13:45
* fanout 消息接收
*/
public class ReceiveLogs01 {
public static final String EXCHANGE_NAME = "logs";
public static void main(String[] args) throws Exception {
//创建信道
Channel channel = RabbitMqUtil.getChannel();
//声明一个交换机
channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
/**
* 生成临时队列
* 当消费者断开与队列的连接的时候，队列就会自动删除
* */
String queueName = channel.queueDeclare().getQueue();
//绑定交换机和队列
channel.queueBind(queueName,EXCHANGE_NAME,"");
System.out.println("receive01等待接收消息，打印消息到控制台.......");
DeliverCallback deliverCallback = (consumerTag, message) -> {
System.out.println(new String(message.getBody()));
};
channel.basicConsume(queueName,true,deliverCallback,(a)->{});
}
}
将消息根据routingKey发送到指定队列，一个队列可以绑定多个routingKey
消费者
package com.rabbit.review.direct;
import com.rabbit.review.util.RabbitMqUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.Delivery;
/**
* @author zwj
* @date 2022/2/13 - 14:01
* direct 路由 消费者
*/
public class DirectReceive01 {
private static final String EXCHANGE_NAME = "direct_logs";
private static final String ERROR_QUEUE = "error";
private static final String INFO_QUEUE = "info";
private static final String WARNING_QUEUE = "warning";
public static void main(String[] args) throws Exception {
//创建信道
Channel channel = RabbitMqUtil.getChannel();
//声明交换机
channel.exchangeDeclare(EXCHANGE_NAME,"direct");
//声明队列
channel.queueDeclare(ERROR_QUEUE,false,false,false,null);
channel.queueDeclare(INFO_QUEUE,false,false,false,null);
channel.queueDeclare(WARNING_QUEUE,false,false,false,null);
//绑定交换机和队列
channel.queueBind(ERROR_QUEUE,EXCHANGE_NAME,"error");
channel.queueBind(INFO_QUEUE,EXCHANGE_NAME,"info");
channel.queueBind(WARNING_QUEUE,EXCHANGE_NAME,"warning");
System.out.println("等待接收消息");
//error队列接收消息
DeliverCallback deliverCallbackError = (consumerTag, message) -> {
System.out.println("error接收到消息:"+new String(message.getBody()));
};
channel.basicConsume(ERROR_QUEUE,true,deliverCallbackError,a -> {});
//info队列接收消息
DeliverCallback deliverCallbackInfo = (consumerTag, message) -> {
System.out.println("info接收到消息:"+new String(message.getBody()));
};
channel.basicConsume(INFO_QUEUE,true,deliverCallbackInfo,a -> {});
//warning队列接收消息
DeliverCallback deliverCallbackWarning = (consumerTag, message) -> {
System.out.println("warning接收到消息:"+new String(message.getBody()));
};
channel.basicConsume(WARNING_QUEUE,true,deliverCallbackWarning,a -> {});
}
}
生产者
package com.rabbit.review.direct;
import com.rabbit.review.util.RabbitMqUtil;
import com.rabbitmq.client.Channel;
import java.nio.charset.StandardCharsets;
import java.util.Scanner;
/**
* @author zwj
* @date 2022/2/13 - 14:12
* direct 路由 生产者
*/
public class DirectSend {
private static final String EXCHANGE_NAME = "direct_logs";
public static void main(String[] args) throws Exception {
//创建信道
Channel channel = RabbitMqUtil.getChannel();
//声明交换机
channel.exchangeDeclare(EXCHANGE_NAME,"direct");
//发送消息
Scanner scanner = new Scanner(System.in);
String message = "text article";
while (scanner.hasNext()){
String type = scanner.next();
System.out.println(type);
switch (type){
case "error":
channel.basicPublish(EXCHANGE_NAME,"error",null,message.getBytes(StandardCharsets.UTF_8));
break;
case "info":
channel.basicPublish(EXCHANGE_NAME,"info",null,message.getBytes(StandardCharsets.UTF_8));
break;
case "warning":
channel.basicPublish(EXCHANGE_NAME,"warning",null,message.getBytes(StandardCharsets.UTF_8));
break;
default:
System.out.println("没有此队列");
break;
}
}
}
}
当队列很多的时候，消息需要交叉发送，路由模式就不好实现了,使用路由模式，一个队列有多个routingkey，不好维护,所以出现了主题模式
topic的routingKey不能随便写，必须是单词列表，多个单词用点隔开，最多不能超过255字节
text.info.error
text.info.warning
// 在上面图中可以这样来表示
b1 a.b.c
b2 a.b.d
// 或者queue2确定都会被访问，可以使用一条routingKey,用通配符表示
b3 a.*.*
// 通配符
* 表示任意一个单词
# 表示任意多个单词
消费者
package com.rabbit.review.topic;
import com.rabbit.review.util.RabbitMqUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
/**
* @author zwj
* @date 2022/2/13 - 14:45
* topic 主题模式 消费者
*/
public class TopicReceive {
private static final String EXCHANGE_NAME = "topic_log";
public static void main(String[] args) throws Exception {
//创建信道
Channel channel = RabbitMqUtil.getChannel();
//声明交换机
channel.exchangeDeclare(EXCHANGE_NAME,"topic");
//声明队列
channel.queueDeclare("Q1",false,false,false,null);
channel.queueDeclare("Q2",false,false,false,null);
//绑定交换机和队列
channel.queueBind("Q1",EXCHANGE_NAME,"*.orange.*");
channel.queueBind("Q2",EXCHANGE_NAME,"*.*.rabbit");
channel.queueBind("Q2",EXCHANGE_NAME,"lazy.#");
System.out.println("等待接收消息");
//Q1
DeliverCallback deliverCallback1 = (consumerTag, message) -> {
System.out.println("Q1接收消息:"+new String(message.getBody(),"UTF-8")+"; 键:"+message.getEnvelope().getRoutingKey());
};
channel.basicConsume("Q1",true,deliverCallback1,consumerTag -> {});
//Q2
DeliverCallback deliverCallback2 = (consumerTag, message) -> {
System.out.println("Q2接收消息:"+new String(message.getBody(),"UTF-8")+"; 键:"+message.getEnvelope().getRoutingKey());
};
channel.basicConsume("Q2",true,deliverCallback2,consumerTag -> {});
}
}
生产者
package com.rabbit.review.topic;
import com.rabbitmq.client.Channel;
import com.zwj.rabbitmq.util.RabbitMqUtils;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
/**
* @author zwj
* @date 2022/2/13 - 14:52
* topic 生产者
*/
public class TopicSend {
//交换机名称
public static final String EXCHANGE_NAME = "topic_log";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtils.getChannel();
channel.exchangeDeclare(EXCHANGE_NAME,"topic");
Map<String,String> bindingMap = new HashMap<>();
bindingMap.put("quick.orange.rabbit","被队列Q1Q2接收到");
bindingMap.put("lazy.orange.element","被队列Q1Q2接收到");
bindingMap.put("quick.orange.fox","被队列Q1接收到");
bindingMap.put("lazy.brown.fox","被队列Q2接收到");
bindingMap.put("lazy.pink.rabbit","被队列Q2接收到");
bindingMap.put("quick.brown.fox","丢弃");
bindingMap.put("quick.orange.male.rabbit","丢弃");
bindingMap.put("lazy.orange.male.rabbit","被队列Q2接收到");
for (Map.Entry<String, String> stringEntry : bindingMap.entrySet()) {
String routingKey = stringEntry.getKey();
String message = stringEntry.getValue();
channel.basicPublish(EXCHANGE_NAME,routingKey,null,message.getBytes(StandardCharsets.UTF_8));
System.out.println("生产者发送键:"+routingKey+"; 生产者发送消息:"+message);
}
}
}
为了防止消息丢失，rabbitMq引入消息应答机制，消费者在接收消息并处理消息后，告诉rabbitMq，rabbitMq才可以删除
消息一旦到达，立刻对rabbitMq做出应答，有可能还没有操作，rabbitMq就将通道关闭了,吞吐量提高，安全性降低
1.Channel.basicAck() # 肯定确认,告诉rabbitMq可以删除消息了
2.Channel.basicNack() # 否定确认
3.Channel.basicReject() # 否定确认 比Nack少一个批量处理参数，不处理当前消息了，可以丢弃了
true:将信道上所有的消息都进行应答，包括等待中的
false:只应答当前操作的消息
如果消费者由于某些原因失去连接，导致消息未被处理完毕，rabbitMq了解到后，会将消息交给其他还存活的消费者处理，会把消费失败的消息重新添加到队列的尾端
#可以设置basicNack第三个参数控制是否开启,默认开启
basicNack(long tags,multi = false, requeue = true)
消费者代码
package com.rabbit.review.handAck;
import com.rabbit.review.util.RabbitMqUtil;
import com.rabbit.review.util.SleepUtil;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import java.nio.charset.StandardCharsets;
/**
* @author zwj
* @date 2022/2/12 - 16:25
* 手动应答工作线程
*/
public class HandAckWorker {
//队列名称
public static final String ACK_QUEUE_NAME = "ack_queue";
public static void main(String[] args) throws Exception{
//工具类创建信道
Channel channel = RabbitMqUtil.getChannel();
//消息接收
DeliverCallback deliverCallback = (consumerTag, message) -> {
SleepUtil.sleep(30);
System.out.println("ackSlowly接收到消息:"+new String(message.getBody(), StandardCharsets.UTF_8));
/**
* 参数
* 1.消息的标记，就像数组下标之类的
* 2.是否批量处理
* */
channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
};
//消息取消消费
CancelCallback cancelCallback = (consumerTag) -> {
System.out.println(consumerTag+"消息被取消消费");
};
//消息应答队列
channel.basicConsume(ACK_QUEUE_NAME,false,deliverCallback,cancelCallback);
}
}
生产者
package com.rabbit.review.handAck;
import com.rabbit.review.util.RabbitMqUtil;
import com.rabbitmq.client.Channel;
import java.nio.charset.StandardCharsets;
import java.util.Scanner;
/**
* @author zwj
* @date 2022/2/12 - 16:29
* 手动应答生产者
*/
public class AckTask {
//队列名称
public static final String ACK_QUEUE_NAME = "ack_queue";
public static void main(String[] args) throws Exception {
//获取信道
Channel channel = RabbitMqUtil.getChannel();
//声明队列
channel.queueDeclare(ACK_QUEUE_NAME,false,false,false,null);
//发送消息
Scanner scanner = new Scanner(System.in);
while (scanner.hasNext()){
String message = scanner.next();
channel.basicPublish("",ACK_QUEUE_NAME,null,message.getBytes(StandardCharsets.UTF_8));
System.out.println("生产者发送消息:"+message);
}
}
}
防止RabbitMQ崩溃时队列或消息丢失
队列持久化 之前要确定RabbitMQ中没有同名却不持久化的队列，防止造成冲突
channel error; protocol method: #method<channel.close>(reply-code=406,
reply-text=PRECONDITION_FAILED - inequivalent arg 'durable' for queue 'ack_queue' in vhost '/':
received 'true' but current is 'false', class-id=50, method-id=10)
# 队列持久化
channel.queueDeclare(ACK_QUEUE_NAME,durable = true,false,false,null);
# 消息持久化 MessageProPerties.PERSISTENT_TEXT_PLAIN
# 但是并不能保证消息不会丢失，更强有力的持久化，需要使用发布确认
channel.basicPublish("",ACK_QUEUE_NAME,MessageProPerties.PERSISTENT_TEXT_PLAIN,message.getBytes(StandardCharsets.UTF_8));
根据消费者处理速度，合理分配消息
//设置参数为1，实现不公平分发；默认为0轮询分发
//不公平分发,前提是手动应答，在消费端设置
channel.basicQos(1)
规定一次拿几条消息消费,放在自己的ready中等待，处理完一条，就可以再接收一条
经过实验，预取值的设置属于轮询+不公平分发
如果预取值小的处理速度快，就会不停的在队列中拿数据，和预取值长的消费者进行轮询
//设置参数为>1，实现不公平分发+预取值；默认为0轮询分发
//不公平分发,前提是手动应答，在消费端设置
channel.basicQos(2)
开启一个异步线程监听消息持久化的确认信息，只有持久化成功后，才会通知消费者ack
生产者将Channel设置成confirm模式，一旦Channel进入confirm模式，所有在该Channel上面发布的消息都将会被指派一个唯一的ID(从1开始)，一旦消息被投递到所有匹配的队列之后，broker就会发送一个确认给生产者(包含消息的唯一ID)，这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出，broker回传给生产者的确认消息中delivery-tag域包含了确认消息的序列号，此外broker也可以设置basic.ack的multiple域，表示到这个序列号之前的所有消息都已经得到了处理；
confirm模式最大的好处在于他是异步的，一旦发布一条消息，生产者应用程序就可以在等Channel返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果RabbitMQ因为自身内部错误导致消息丢失，就会发送一条nack消息，生产者应用程序同样可以在回调方法中处理该nack消息；
# 开启确认发布
channel.confirmSelect();
# 同步接收确认
boolean flag = channel.waitForConfirms();
异步接收确认
//异步发布确认
public static void publishMessageAsync() throws Exception{
Channel channel = RabbitMqUtils.getChannel();
String queueName = UUID.randomUUID().toString();
channel.queueDeclare(queueName,true,false,false,null);
//开启确认发布
channel.confirmSelect();
/**
* 线程安全有序的一个哈希表 适用于高并发的情况下
*1.轻松的将序号和消息进行关联map
*2.轻松的批量删除条目 只要给到序号
* 3.支持高并发
* */
ConcurrentSkipListMap<Long,String> outStandingConfirms = new ConcurrentSkipListMap<>();
//准备消息的监听器 监听哪些消息成功了，哪些消息失败了
//消息成功回调函数
/**
* 参数
* 1.消息的标记
* 2.是否为批量
* */
ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
System.out.println("确认的消息："+deliveryTag);
if (multiple){
//删除已经确认的消息，剩下的就是未确认的消息
ConcurrentNavigableMap<Long, String> confirmed = outStandingConfirms.headMap(deliveryTag);
confirmed.clear();
}else {
outStandingConfirms.remove(deliveryTag);
}
};
//消息失败回调函数
ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
String message = outStandingConfirms.get(deliveryTag);
System.out.println("未确认的消息："+message);
};
/**
* 1.监听哪些消息成功了
* 2.监听哪些消息失败了
* */
channel.addConfirmListener(ackCallback,nackCallback); //异步监听
//开始时间
long beginTime = System.currentTimeMillis();
//批量发送消息
for (int i = 0;i<MESSAGE_COUNT;i++){
String message = i+"";
channel.basicPublish("",queueName,null,message.getBytes(StandardCharsets.UTF_8));
//记录所有消息
outStandingConfirms.put(channel.getNextPublishSeqNo(),message);
}
//结束时间
long endTime = System.currentTimeMillis();
System.out.println("发布"+MESSAGE_COUNT+"个批量消息共用时:"+(endTime-beginTime));
}
无法被消费的消息，生产者将消息投递到队列中，消费者拿到消息后，由于某些消息无法被消费，导致消息成为了死信，就有了死信队列，为了防止消息数据丢失
# 应用场景
为了保证订单业务的消息数据不丢失，或者用户在商城下单成功并点击去支付后在指定时间未支付时自动失效
# 进入死信的条件
消息被拒绝
消息TTL过期
队列达到最大长度
//生产者设置过期时间
AMQP.BasicProperties properties = new
AMQP.BasicProperties().builder().expiration("10000").build();
// 发送消息
channel.basicPublish(NORMAL_EXCHANGE_NAME,"normal",properties,message.getBytes(StandardCharsets.UTF_8));
普通消费者 设置 死信队列 使用参数
package com.rabbit.review.deadqueue;
import com.rabbit.review.util.RabbitMqUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.Delivery;
import java.util.HashMap;
import java.util.Map;
/**
* @author zwj
* @date 2022/2/13 - 16:47
* 死信队列 正常消费者
*/
public class DeadConsumer01 {
//普通交换机名称
private static final String NORMAL_EXCHANGE_NAME = "normal_exchange";
//死信交换机名称
private static final String DEAD_EXCHANGE_NAME = "dead_exchange";
//普通队列名称
private static final String NORMAL_QUEUE = "normal_queue";
//死信队列名称
private static final String DEAD_QUEUE = "dead_queue";
public static void main(String[] args) throws Exception {
//获取信道
Channel channel = RabbitMqUtil.getChannel();
//声明交换机
channel.exchangeDeclare(NORMAL_EXCHANGE_NAME, "direct");
channel.exchangeDeclare(DEAD_EXCHANGE_NAME, "direct");
//设置普通队列参数
Map<String,Object> arguments = new HashMap<>();
//设置过期时间,一般在生产者设置
// map.put("x-message-ttl",10000);
//设置死信交换机
arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE_NAME);
//设置死信队列 routingKey
arguments.put("x-dead-letter-routing-key","dead");
//设置最大长度
// arguments.put("x-max-length",6);
//声明队列
channel.queueDeclare(NORMAL_QUEUE, false, false, false, arguments);
channel.queueDeclare(DEAD_QUEUE, false, false, false, null);
//绑定交换机和队列
channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE_NAME,"normal");
channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE_NAME,"dead");
System.out.println("等待接收消息......");
//接收消息
DeliverCallback deliverCallback = (consumerTag, message) -> {
//拒绝消息
String msg = new String(message.getBody(), "UTF-8");
if (msg.equals("info5")){
System.out.println("consumer1接收的消息是："+new String(message.getBody(),"UTF-8")+"被拒绝");
channel.basicReject(message.getEnvelope().getDeliveryTag(),false);
}else {
System.out.println("consumer1接收的消息是："+new String(message.getBody(),"UTF-8"));
channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
}
};
//测试拒绝消息，必须开启手动应答
channel.basicConsume(NORMAL_QUEUE,false,deliverCallback,a->{});
}
}
生产者
package com.rabbit.review.deadqueue;
import com.rabbit.review.util.RabbitMqUtil;
import com.rabbit.review.util.SleepUtil;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import java.nio.charset.StandardCharsets;
/**
* @author zwj
* @date 2022/2/13 - 17:08
*/
public class DeadProduct {
//普通交换机名称
private static final String NORMAL_EXCHANGE_NAME = "normal_exchange";
public static void main(String[] args) throws Exception {
//获取信道
Channel channel = RabbitMqUtil.getChannel();
//声明交换机
channel.exchangeDeclare(NORMAL_EXCHANGE_NAME,"direct");
//设置过期时间
AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();
//发送消息
for (int i = 1; i <= 10; i++) {
String message = "info"+i;
channel.basicPublish(NORMAL_EXCHANGE_NAME,"normal",null,message.getBytes(StandardCharsets.UTF_8));
}
System.out.println("发送成功");
}
}
死信队列消费者
package com.rabbit.review.deadqueue;
import com.rabbit.review.util.RabbitMqUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.Delivery;
/**
* @author zwj
* @date 2022/2/13 - 17:06
* 死信队列消费者
*/
public class DeadConsumer02 {
//死信队列名称
private static final String DEAD_QUEUE = "dead_queue";
public static void main(String[] args) throws Exception {
//获取信道
Channel channel = RabbitMqUtil.getChannel();
//声明队列
// channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
System.out.println("等待接收消息。。。。");
DeliverCallback deliverCallback = (consumerTag, message) -> {
System.out.println("死信队列收到:"+new String(message.getBody()));
};
//接收消息
channel.basicConsume(DEAD_QUEUE,true,deliverCallback,a->{});
}
}
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<!--rabbitmq依赖客户端-->
<!-- rabbit-mq -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.75</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<!-- https://mvnrepository.com/artifact/io.springfox/springfox-swagger2 -->
<dependency>
<groupId>io.springfox</groupId>
<artifactId>springfox-swagger2</artifactId>
<version>2.9.2</version>
</dependency>
<dependency>
<groupId>io.springfox</groupId>
<artifactId>springfox-swagger-ui</artifactId>
<version>2.9.2</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.springframework.amqp/spring-rabbit-test -->
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit-test</artifactId>
<version>2.4.1</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
应用场景
1.限时订单
2.会议定时提醒
延迟队列配置类 ， 将队列需要的各种对象注入到spring容器中
package com.zwj.springbootrabbitmq.config;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;
/**
* @author zwj
* @date 2022/1/10 - 17:19
* ttl队列 配置文件类
*/
@Configuration
public class TtlQueueConfig {
//普通交换机的名称
public static final String X_EXCHANGE = "X";
//死信交换机名称
public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
//普通队列名称
public static final String QUEUE_A = "QA";
public static final String QUEUE_B = "QB";
//通用普通队列
public static final String QUEUE_C = "QC";
//死信队列名称
public static final String DEAD_LETTER_QUEUE = "QD";
//声明xExchange
@Bean("xExchange")
public DirectExchange xExchange(){
return new DirectExchange(X_EXCHANGE);
}
//声明yExchange
@Bean("yExchange")
public DirectExchange yExchange(){
return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
}
//声明普通队列
@Bean("queueA")
public Queue queueA(){
Map<String,Object> arguments = new HashMap<>(3);
//设置死信交换机
arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);
//设置死信routerKey
arguments.put("x-dead-letter-routing-key","YD");
//设置TTL 单位是ms
arguments.put("x-message-ttl",10000);
return QueueBuilder.durable(QUEUE_A).withArguments(arguments).build();
}
@Bean("queueB")
public Queue queueB(){
Map<String,Object> arguments = new HashMap<>(3);
//设置死信交换机
arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);
//设置死信routerKey
arguments.put("x-dead-letter-routing-key","YD");
//设置TTL 单位是ms
arguments.put("x-message-ttl",40000);
return QueueBuilder.durable(QUEUE_B).withArguments(arguments).build();
}
@Bean("queueC")
public Queue queueC(){
Map<String,Object> arguments = new HashMap<>(2);
//设置死信交换机
arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);
//设置死信routerKey
arguments.put("x-dead-letter-routing-key","YD");
return QueueBuilder.durable(QUEUE_C).withArguments(arguments).build();
}
//声明死信队列
@Bean("queueD")
public Queue queueD(){
return QueueBuilder.durable(DEAD_LETTER_QUEUE).build();
}
@Bean
public Binding queueABindingX(@Qualifier("queueA") Queue queueA,
@Qualifier("xExchange") DirectExchange xExchange){
return BindingBuilder.bind(queueA).to(xExchange).with("XA");
}
@Bean
public Binding queueBBindingX(@Qualifier("queueB") Queue queueB,
@Qualifier("xExchange") DirectExchange xExchange){
return BindingBuilder.bind(queueB).to(xExchange).with("XB");
}
@Bean
public Binding queueCBindingX(@Qualifier("queueC") Queue queueC,
@Qualifier("xExchange") DirectExchange xExchange){
return BindingBuilder.bind(queueC).to(xExchange).with("XC");
}
@Bean
public Binding queueDBindingX(@Qualifier("queueD") Queue queueD,
@Qualifier("yExchange") DirectExchange yExchange){
return BindingBuilder.bind(queueD).to(yExchange).with("YD");
}
}
生产者
package com.zwj.springbootrabbitmq.controller;
import com.zwj.springbootrabbitmq.config.DelayedQueueConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.Date;
/**
* @author zwj
* @date 2022/1/10 - 17:36
* 发送延迟消息
*/
@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMsgController {
@Autowired
private RabbitTemplate rabbitTemplate;
//开始发消息
@GetMapping("/sendMsg/{message}/{time}")
public void sendMsg(@PathVariable String message){
log.info("当前时间：{},发送一条信息给两个TTL队列：{}",new Date().toString(),message);
rabbitTemplate.convertAndSend("X","XA","消息来自ttl为10s的队列:"+message);
rabbitTemplate.convertAndSend("X","XB","消息来自ttl为40s的队列:"+message);
}
//生产者设置队列延迟时长
@GetMapping("/sendExpirationMsg/{message}/{time}")
public void sendMsg(@PathVariable String message,@PathVariable int time){
log.info("当前时间：{},发送一条信息给两个TTL队列：{}",new Date().toString(),message);
rabbitTemplate.convertAndSend("X","XC","消息来自ttl为"+time+"s的队列："+message,msg -> {
//发送消息的时长
msg.getMessageProperties().setExpiration(String.valueOf(time*1000));
return msg;
});
}
}
死信队列消费者
package com.zwj.springbootrabbitmq.consumer;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Date;
/**
* @author zwj
* @date 2022/1/10 - 17:43
* 队列TTL 消费者
*/
@Component
@Slf4j
public class DeadLetterQueueConsumer {
//接收消息
@RabbitListener(queues = "QD")
public void receiveD(Message message, Channel channel) throws Exception{
String msg = new String(message.getBody());
log.info("当前时间：{},收到死信队列的消息:{}",new Date().toString(),msg);
}
}
RabbitMQ默认只会一个一个的检查是否过期，第一条没被检测完成之前，第二条不会被检测，使用RabbitMQ插件实现延迟队列
使用交换机进行延迟,不需要设置死信队列，只需要一个延迟交换机和一个队列，交换机延迟后，将消息发送到队列
https://www.rabbitmq.com/community-plugins.html
rabbitmq_delayed_message_exchange
基于插件的延迟交换机，将数据保存到分布式表中进行延迟，时间到了将信息发布到队列
解决ttl队列不能按时间发布消息的缺点，ttl队列必须等第一个消息延迟完毕，才能进行第二个消息
插件安装在rabbitMQ的plugins文件夹下
/usr/lib/rabbitmq/lib/rabbitmq_server-3.8.8/plugins
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
重启rabbitmq
systemctl restart rabbitmq-server
rabbitMQ配置类
package com.zwj.springbootrabbitmq.config;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.CustomExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;
/**
* @author zwj
* @date 2022/1/10 - 18:46
* https://www.rabbitmq.com/community-plugins.html
* rabbitmq_delayed_message_exchange
* 基于插件的延迟交换机，将数据保存到分布式表中进行延迟，时间到了将信息发布到队列
* 解决ttl队列不能按时间发布消息的缺点，ttl队列必须等第一个消息延迟完毕，才能进行第二个消息
*/
@Configuration
public class DelayedQueueConfig {
//队列
public static final String DELAYED_QUEUE_NAME = "delayed.queue";
//交换机
public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange";
//routingKey
public static final String DELAYED_ROUTING_KEY = "delayed.routingkey";
@Bean
public Queue delayedQueue(){
return new Queue(DELAYED_QUEUE_NAME);
}
//声明交换机,自定义交换机
@Bean
public CustomExchange delayedExchange(){
Map<String,Object> arguments = new HashMap<>();
//检测消息延迟时间，如果达到投递时间，通过x-delayed-type 标记的交换机类型进行投递
arguments.put("x-delayed-type","direct"); //设置延迟类型 ，直接类型 不用分别发布
/**
* 1.交换机的名称
* 2.交换机的类型
* 3.持久化
* 4.是否需要自动删除
* 5.其他参数
* */
return new CustomExchange(DELAYED_EXCHANGE_NAME,"x-delayed-message",true,false,arguments);
}
//绑定
@Bean
public Binding delayedQueueBindingDelayedExchange(@Qualifier("delayedQueue") Queue delayedQueue,
@Qualifier("delayedExchange") CustomExchange delayedExchange){
return BindingBuilder.bind(delayedQueue).to(delayedExchange).with(DELAYED_ROUTING_KEY).noargs();
}
}
生产者
//开始发消息 基于插件的 消息 及延迟的时间
@GetMapping("/sendDelayMsg/{message}/{delayTime}")
public void sendMsg(@PathVariable String message,@PathVariable Integer delayTime){
log.info("当前时间：{},发送一条时长{}毫秒的信息给延迟队列：{}",new Date().toString(),delayTime,message);
rabbitTemplate.convertAndSend(DelayedQueueConfig.DELAYED_EXCHANGE_NAME, DelayedQueueConfig.DELAYED_ROUTING_KEY,
message,msg -> {
//发送消息的时候 延迟时长 单位 ms
msg.getMessageProperties().setDelay(delayTime);
return msg;
});
}
消费者监听
package com.zwj.springbootrabbitmq.consumer;
import com.zwj.springbootrabbitmq.config.DelayedQueueConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Date;
/**
* @author zwj
* @date 2022/1/10 - 19:07
* 消费者 基于插件的延迟
*/
@Slf4j
@Component
public class DelayQueueConsumer {
//监听消息
@RabbitListener(queues = DelayedQueueConfig.DELAYED_QUEUE_NAME)
public void receiveDelayQueue(Message message){
String msg = new String(message.getBody());
log.info("当前时间：{}，收到延迟队列的消息：{}",new Date().toString(),msg);
}
}
交换机确认收到消息
# 配置类
# 开启确认发布 启用交换机确认回调接口 none 默认不开启
spring.rabbitmq.publisher-confirm-type = correlated
参数
currelated 发布消息成功到交换器后回触发回调方法
simple
两种效果
1.和correlated一样回触发回调方法
2.发布消息成功后使用rabbitTemplat调用waitForConfirms或waitForConfirmsOrDie方法等待broker节点返回发送结果，根据结果来判定下一步的逻辑，要注意的是，waitForConfirmsOrDie方法如果返回false则会关闭channel,则接下来无法发送消息到broker
Mandatory参数
在仅开启了生产者确认机制的情况下，交换机接收到消息后，会直接给消息生产者发送确认消息，如果发现该消息不可路由，那么消息会被直接丢弃，此时生产者是不知道的，
所以通过mandatory参数将不可达目的的消息返回给生产者
#开启发布回退
spring.rabbitmq.publisher-returns=true
在使用 RabbitMQ 的时候，作为消息发送方希望杜绝任何消息丢失或者投递失败场景。 RabbitMQ 为我们提供了两种方式用来控制消息的投递可靠性模式。
-confirm 确认模式
-return 退回模式
rabbitmq 整个消息投递的路径为：
producer—>rabbitmq broker—>exchange—>queue—>consumer
-消息从 producer 到 exchange 则会返回一个 confirmCallback 。
-消息从 exchange–>queue 投递失败则会返回一个 returnCallback 。
我们将利用这两个 callback 控制消息的可靠性投递
rabbitMQ配置类
package com.zwj.springbootrabbitmq.config;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @author zwj
* @date 2022/1/10 - 20:12
* 发布确认高级
*/
@Configuration
public class ConfirmConfig {
//交换机
public static final String CONFIRM_EXCHANGE_NAME = "confirm_exchange";
//队列
public static final String CONFIRM_QUEUE_NAME = "confirm_queue";
//routingKey
public static final String CONFIRM_ROUTING_KEY = "key1";
/**
* */
//备份交换机
public static final String BACKUP_EXCHANGE_NAME = "backup_exchange";
//备份队列
public static final String BACKUP_QUEUE_NAME = "backup_queue";
//报警队列
public static final String WARNING_QUEUE_NAME = "warning_queue";
//声明交换机
@Bean
public DirectExchange confirmExchange(){
return new DirectExchange(CONFIRM_EXCHANGE_NAME);
}
//声明队列
@Bean
public Queue confirmQueue(){
return new Queue(CONFIRM_QUEUE_NAME);
}
//绑定
@Bean
public Binding queueBindingExchange(@Qualifier("confirmQueue") Queue confirmQueue,
@Qualifier("confirmExchange") DirectExchange confirmExchange){
return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(CONFIRM_ROUTING_KEY);
}
}
确认发布回调方法
package com.zwj.springbootrabbitmq.config;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
/**
* @author zwj
* @date 2022/1/10 - 20:32
*/
@Slf4j
@Component
public class MyCallBack implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback{
@Autowired
private RabbitTemplate rabbitTemplate;
//注入 将实现类set到RabbitTemplate中的ConfirmCallback参数中
@PostConstruct //对象加载完依赖注入后执行
public void init(){
//注入
rabbitTemplate.setConfirmCallback(this);
rabbitTemplate.setReturnsCallback(this);
}
/**
* 交换机回调方法
* 1.发消息 交换机接收到了 回调
* 1.1 correlationData 保存回调信息的ID及相关信息
* 1.2 交换机收到信息 ack = true
* 1.3 cause null
* 2.发消息 交换机接收失败了 回调
* 2.1 correlationData 保存回调信息的ID及相关信息
* 2.2 交换机收到消息 ack = false
* 2.3 cause 失败的原因
* */
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
String id = correlationData != null ? correlationData.getId() : "";
if (ack){
log.info("交换机已经收到了id为：{}的消息",id);
}else {
log.info("交换机还未收到id为：{}的消息，由于原因：{}",id,cause);
}
}
//只有在当消息传递过程中不可达目的地是将消息返回给生产者
//只有不可达目的地的时候才进行回退
//消息回退
@Override
public void returnedMessage(ReturnedMessage returned) {
log.error("消息{},被交换机{}退回，退回原因:{},路由key:{}",
new String(returned.getMessage().getBody()),returned.getExchange(),returned.getReplyText(),
returned.getRoutingKey());
//重试
rabbitTemplate.convertAndSend(returned.getExchange(),returned.getRoutingKey(),
returned.getBody())
}
}
生产者
package com.zwj.springbootrabbitmq.controller;
import com.zwj.springbootrabbitmq.config.ConfirmConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
/**
* @author zwj
* @date 2022/1/10 - 20:19
* //开始发消息 测试确认
*/
@Slf4j
@RestController
@RequestMapping("/confirm")
public class ProducerController {
@Autowired
RabbitTemplate rabbitTemplate;
//发消息
@GetMapping("/sendMessage/{message}")
public void sendMessage(@PathVariable String message){
//回调信息ID 回调方法需要使用的信息
CorrelationData correlationData = new CorrelationData("1");
rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME,ConfirmConfig.CONFIRM_ROUTING_KEY+2,message,correlationData);
log.info("发送消息内容为：{}",message);
}
}
alternate-exchange
package com.zwj.springbootrabbitmq.config;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @author zwj
* @date 2022/1/10 - 20:12
* 发布确认高级
*/
@Configuration
public class ConfirmConfig {
//交换机
public static final String CONFIRM_EXCHANGE_NAME = "confirm_exchange";
//队列
public static final String CONFIRM_QUEUE_NAME = "confirm_queue";
//routingKey
public static final String CONFIRM_ROUTING_KEY = "key1";
/**
* */
//备份交换机
public static final String BACKUP_EXCHANGE_NAME = "backup_exchange";
//备份队列
public static final String BACKUP_QUEUE_NAME = "backup_queue";
//报警队列
public static final String WARNING_QUEUE_NAME = "warning_queue";
//声明交换机
@Bean
public DirectExchange confirmExchange(){
// return new DirectExchange(CONFIRM_EXCHANGE_NAME);
//转向其他交换机
return ExchangeBuilder.directExchange(CONFIRM_EXCHANGE_NAME).
withArgument("alternate-exchange",BACKUP_EXCHANGE_NAME).build();
}
//声明队列
@Bean
public Queue confirmQueue(){
return new Queue(CONFIRM_QUEUE_NAME);
}
//绑定
@Bean
public Binding queueBindingExchange(@Qualifier("confirmQueue") Queue confirmQueue,
@Qualifier("confirmExchange") DirectExchange confirmExchange){
return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(CONFIRM_ROUTING_KEY);
}
//声明备份交换机
@Bean
public FanoutExchange backupExchange(){
return new FanoutExchange(BACKUP_EXCHANGE_NAME);
}
//声明备份队列
@Bean
public Queue backupQueue(){
return new Queue(BACKUP_QUEUE_NAME);
}
//声明报警队列
@Bean
public Queue warningQueue(){
return new Queue(WARNING_QUEUE_NAME);
}
//绑定
@Bean
public Binding backupQueueBindingBackupExchange(@Qualifier("backupQueue") Queue backupQueue,
@Qualifier("backupExchange") FanoutExchange backupExchange){
return BindingBuilder.bind(backupQueue).to(backupExchange);
}
@Bean
public Binding warningQueueBindingBackupExchange(@Qualifier("warningQueue") Queue warningQueue,
@Qualifier("backupExchange") FanoutExchange backupExchange){
return BindingBuilder.bind(warningQueue).to(backupExchange);
}
}
== 概念 ==
用户对于同一操作发起的一次请求或多次请求的结果是一致的，不会因为多次点击而产生副作用
比如支付，用户支付后，返回结果时网络异常，此时钱已经扣了，用户第二次点击，进行第二次扣款，返回结果成功
以前的单应用系统中，我们只需要把数据操作放入事务中即可，发生错误立即回滚
消费者在消费MQ中的消息时，MQ已经将消息发送给消费者，消费者在给MQ返回ack时网络中断，故MQ未收到确认消息，该条消息会重新发给其他的消费者，或者在网络重连后发送给消费者，但实际上该消费者已成功消费了该条消息，造成消费者消费了重复的消息
解决思路
MQ消费者的幂等性的解决一般使用全局ID或者写个唯一标识比如时间戳，或者UUID或者订单消费者消费MQ中的消息也可利用MQ的该id来判断，或者可按自己的规则生成一个全局id,每次消费消息时用该id先判断该消息是否已消费过
消费端的幂等性保障
重复提交保证幂等性的常用两种操作
1.唯一ID+指纹码机制，利用数据库主键去重
指纹码：我们的一些规则或者时间戳加别的服务给到的唯一信息码，它并不一定时系统生成的，基本都是由我们的业务规则拼接而来，但是一定要保证唯一性，然后就利用查询 语句进行判断这个id是否存在数据库中，优势是简单拼接就可以查重，劣势是高并发下，性能差
2.利用redis原子性去实现
利用redis执行setnx，天然具有幂等性，从而实现不重复消费
订单催付
天猫或淘宝及时将订单推送给我们，如果用户在设定时间内未付款就会给用户推送一条短信提醒
但是商家对于淘宝天猫来说分为大商家和小商家，大商家的订单要优先处理
redis可以做定时轮询，但是没有优先级
所以可以用rabbitmq设置优先级
优先级队列(0-255)越大越优先执行
注意： 要让队列实现优先级需要做的事情
队列需要设置为优先级队列，消息需要设置消息的优先级，消费者需要等待消息已经发送到队列中才去消费，因为这样才有机会进行排序
//设置优先级
Map<String,Object> arguments = new HashMap<>();
arguments.put("x-max-priority",10); //不要设置过大，浪费内存
channel.queueDeclare(QUEUE_NAME,false,false,false,arguments);
AMQP.BasicProperties properties = new
AMQP.BasicProperties().builder().priority(5).build();
channel.basicPublish("",QUEUE_NAME,properties,message.getBytes(StandardCharsets.UTF_8));
// springboot消息设置优先级 队列在配置文件中设置
rabbitTemplate.convertAndSend("X","XC","消息来自ttl为"+time+"s的队列："+message,msg -> {
//发送消息的优先级
msg.getMessageProperties().setPriority();
return msg;
});
正常情况：消息保存在内存上
惰性队列：消息保存在磁盘上
应用场景
数据量大，消费者宕机，存储在磁盘上
args.put("x-queue-mode","lazy")
修改名称
1.修改3台电脑的主机名称
vim /etc/hostname
2.配置各个节点的hosts文件，让各个节点都能互相识别对方 每个节点都写
vim /etc/hosts
ip1 node1
ip2 node2
ip3 node3
3.以确保各个节点的cookie文件使用的是同一个值
在node1上执行远程操作命令
scp /var/lib/rabbitmq/.erlang.cookie root@node2:/var/lib/rabbitmq/.erlang.cookie
scp /var/lib/rabbitmq/.erlang.cookie root@node3:/var/lib/rabbitmq/.erlang.cookie
4.启动RabbitMQ服务，顺带启动Erlang虚拟机和RabbitMQ应用服务(在三台节点上分别执行)
rabbitmq-server -detached
5.在节点2执行
rabbitmqctl stop_app
(rabbitmqctl stop 会关闭erlang虚拟机 ，rabbitmqctl stop_app只关闭rabbitmq服务)
rabbitmqctl reset
rabbitmqctl join_cluster rabbit@node1
rabbitmqctl start_app(之启动应用服务)
6.在节点3执行
rabbitmqctl stop_app
(rabbitmqctl stop 会关闭erlang虚拟机 ，rabbitmqctl stop_app只关闭rabbitmq服务)
rabbitmqctl reset
rabbitmqctl join_cluster rabbit@node2
rabbitmqctl start_app(之启动应用服务)
7.集群状态
rabbitmq cluster_status
8.需要重新设置用户
创建账号
rabbitmqctl add_user admin 123
设置用户角色
rabbitmqctl set_user_tags admin administrator
设置用户权限
rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"
9.解除集群节点(节点2和节点3 都执行)
rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl start_app
rabbitmqctl cluster_status //检查状态
rabbitmqctl forget_cluster_node rabbit@node2 (node1上执行)