Lors de l'utilisation de RabbitMQ, en tant qu'expéditeur du message, nous espérons éviter toute perte ou livraison de message scène d'échec. RabbitMQ nous offre deux façons de contrôler le mode de fiabilité de livraison des messages.
confirmer le mode de confirmation
retour mode de retour
Si le message est envoyé du producteur à l'échange, un confirmCallback sera renvoyé 🎜#
Nous pouvons utiliser ces deux rappels pour contrôler la fiabilité de la livraison du message
#. 🎜🎜#Mode de confirmation
<!--
* 确认模式：
* 步骤：
* 1. 确认模式开启：ConnectionFactory中开启publisher-confirms="true"
-->
<!-- 定义rabbitmq connectionFactory -->
<rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"
port="${rabbitmq.port}"
username="${rabbitmq.username}"
password="${rabbitmq.password}"
virtual-host="${rabbitmq.virtual-host}"
publisher-confirms="true"/>#🎜🎜 #
Étant donné que le message est envoyé normalement à la file d'attente, la valeur de cause renvoyée est vide. Si une exception se produit, la cause est la raison de l'exception/*
* 确认模式：
* 步骤：
* 2. 在rabbitTemplate定义ConfirmCallBack回调函数
*/
@Test
public void queueTest(){
rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
/**
*
* @param correlationData 相关配置信息
* @param ack exchange交换机 是否成功收到了消息。true 成功，false代表失败
* @param cause 失败原因
*/
System.out.println("confirm方法被执行了....");
if (ack) {
//接收成功
System.out.println("接收成功消息" + cause);
} else {
//接收失败
System.out.println("接收失败消息" + cause);
//做一些处理，让消息再次发送。
}
}
});
//路由键与队列同名
rabbitTemplate.convertAndSend("spring_queue", "message confirm....");
} <!-- 定义rabbitmq connectionFactory -->
<rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"
port="${rabbitmq.port}"
username="${rabbitmq.username}"
password="${rabbitmq.password}"
virtual-host="${rabbitmq.virtual-host}"
publisher-returns="true"/>
# 🎜🎜#Confirmation manuelle : accusé de réception="manuel"
La confirmation automatique signifie qu'une fois le message reçu par le consommateur, il confirmera automatiquement la réception et supprimera le message correspondant du cache de messages de RabbitMQ. Cependant, dans le traitement métier réel, il est très probable que le message soit perdu s'il y a une exception dans le traitement métier après la réception du message. Si la méthode de confirmation manuelle est définie, vous devez appeler Channel.basicAck() une fois le traitement opérationnel réussi et vous connecter manuellement. Si une exception se produit, appelez la méthode Channel.basicNack() pour lui permettre de renvoyer automatiquement le message.
@Test
public void queueTest(){
//1.设置交换机处理失败消息的模式
rabbitTemplate.setMandatory(true);
//2.设置ReturnCallBack
rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
/**
* @param message 消息对象
* @param replyCode 错误码
* @param replyText 错误信息
* @param exchange 交换机
* @param routingKey 路由键
*/
@Override
public void returnedMessage(Message message, int replyCode, String
replyText, String exchange, String routingKey) {
System.out.println("return 执行了....");
System.out.println(message);
System.out.println(replyCode);
System.out.println(replyText);
System.out.println(exchange);
System.out.println(routingKey);
//处理
}
});
//手动添加错误路由模拟错误发生
rabbitTemplate.convertAndSend("spring_topic_exchange", "return123", "return message...");
}Le code de la classe de surveillance est le suivant :
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual"> .....
1. Assurez-vous que le mécanisme d'acquittement est une confirmation manuelle
2 Attribut de configuration de l'écouteur-conteneur perfetch = 1, ce qui signifie que le consommateur extrait un message de mq pour le consommer. à chaque fois jusqu'à Après avoir confirmé manuellement que la consommation est terminée, le message suivant sera envoyé.
public class AckListener implements ChannelAwareMessageListener {
@Override
public void onMessage(Message message, Channel channel) throws Exception {
long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
//1.接收转换消息
System.out.println(new String(message.getBody()));
//2. 处理业务逻辑
System.out.println("处理业务逻辑...");
int i = 3/0;//出现错误
// 3. 手动签收
channel.basicAck(deliveryTag,true);
} catch (Exception e) {
//e.printStackTrace();
//4.拒绝签收
/*
*第三个参数：requeue：重回队列。如果设置为true，则消息重新回到queue，broker会
*重新发送该消息给消费端
*/
channel.basicNack(deliveryTag,true,true);
//channel.basicReject(deliveryTag,true);
}
}
}Producteur, envoie cinq messages
<rabbit:listener-container connection-factory="connectionFactory" auto-declare="true" acknowledge="manual" prefetch="1">
<rabbit:listener ref="topicListenerACK" queue-names="spring_topic_queue_well2"/>
</rabbit:listener-container>Le producteur commente Channel.basicAck(deliveryTag,true) pour ne pas confirmer la réception du message
@Test
public void topicTest(){
/**
* 参数1：交换机名称
* 参数2：路由键名
* 参数3：发送的消息内容
*/
for (int i=0;i<5;i++){
rabbitTemplate.convertAndSend("spring_topic_exchange", "xzk.a", "发送到spring_topic_exchange交换机xzk.cn的消息"+i);
}
}
}
把channel.basicAck(deliveryTag,true)的注释取消掉，即可以自动确认收到消息，重新运行消费者，接收到了另外的四条消息
Time To Live，消息过期时间设置
设置交换机，队列以及队列过期时间为10000ms
<!--ttl-->
<rabbit:queue name="test_queue_ttl" id="test_queue_ttl">
<rabbit:queue-arguments>
<entry key="x-message-ttl" value="10000" value-type="java.lang.Integer"/>
</rabbit:queue-arguments>
</rabbit:queue>
<rabbit:topic-exchange name="test_exchange_ttl">
<rabbit:bindings>
<rabbit:binding pattern="ttl.#" queue="test_queue_ttl"/>
</rabbit:bindings>
</rabbit:topic-exchange>生产者发送10条消息
@Test
public void testTtl() {
for (int i = 0; i < 10; i++) {
rabbitTemplate.convertAndSend("test_exchange_ttl","ttl.hehe","message ttl...");
}
十秒钟后，过期消息消失
设置交换机和队列
<rabbit:queue name="test_queue_ttl" id="test_queue_ttl"/>
<rabbit:topic-exchange name="test_exchange_ttl">
<rabbit:bindings>
<rabbit:binding pattern="ttl.#" queue="test_queue_ttl"/>
</rabbit:bindings>
</rabbit:topic-exchange>生产者发送特定过期消息，用到了MessagePostProcessor这个api
@Test
public void testTtl() {
MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
//1.设置message信息
message.getMessageProperties().setExpiration("5000");//消息的过期时间
//2.返回该消息
return message;
}
};
//消息单独过期
rabbitTemplate.convertAndSend("test_exchange_ttl","ttl.hehe","message ttl...",messagePostProcessor);
}
5s之后
注:
1.如果同时设置队列过期和消息过期，系统会根据哪个过期的时间短而选用哪儿个。
2.设置单独消息过期时，如果该消息不为第一个接受的消息，则不过期。
死信队列，英文缩写：DLX 。Dead Letter Exchange（死信交换机），当消息成为Deadmessage后，可以被重新发送到另一个交换机，这个交换机就是DLX。
消息成为死信的三种情况：
队列消息长度到达限制；
消费者拒接消费消息，basicNack/basicReject,并且不把消息重新放入原目标队列,requeue=false；
原队列存在消息过期设置，消息到达超时时间未被消费；
队列绑定死信交换机：
给队列设置参数： x-dead-letter-exchange 和 x-dead-letter-routing-key
实现
1.声明正常的队列(test_queue_dlx)和交换机(test_exchange_dlx)
<rabbit:queue name="test_queue_dlx" id="test_queue_dlx">
<!--正常队列绑定死信交换机-->
<rabbit:queue-arguments>
<!--x-dead-letter-exchange：死信交换机名称-->
<entry key="x-dead-letter-exchange" value="exchange_dlx" />
<!--3.2 x-dead-letter-routing-key：发送给死信交换机的routingkey-->
<entry key="x-dead-letter-routing-key" value="dlx.hehe" />
<!--4.1 设置队列的过期时间 ttl-->
<entry key="x-message-ttl" value="10000" value-type="java.lang.Integer"/>
<!--4.2 设置队列的长度限制 max-length -->
<entry key="x-max-length" value="10" value-type="java.lang.Integer" />
</rabbit:queue-arguments>
</rabbit:queue>
<rabbit:topic-exchange name="test_exchange_dlx">
<rabbit:bindings>
<rabbit:binding pattern="test.dlx.#" queue="test_queue_dlx">
</rabbit:binding>
</rabbit:bindings>
</rabbit:topic-exchange>2.声明死信队列(queue_dlx)和死信交换机(exchange_dlx)
<rabbit:queue name="queue_dlx" id="queue_dlx"></rabbit:queue>
<rabbit:topic-exchange name="exchange_dlx">
<rabbit:bindings>
<rabbit:binding pattern="dlx.#" queue="queue_dlx"></rabbit:binding>
</rabbit:bindings>
</rabbit:topic-exchange>3.生产端测试
/**
* 发送测试死信消息：
* 1. 过期时间
* 2. 长度限制
* 3. 消息拒收
*/
@Test
public void testDlx(){
//1. 测试过期时间，死信消息
rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息，我会死吗？");
//2. 测试长度限制后，消息死信
/* for (int i = 0; i < 20; i++) {
rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息，我会死吗？");
}*/
//3. 测试消息拒收
//rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息，我会死吗？");
}4.消费端监听
public class DlxListener implements ChannelAwareMessageListener {
@Override
public void onMessage(Message message, Channel channel) throws Exception {
long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
//1.接收转换消息
System.out.println(new String(message.getBody()));
//2. 处理业务逻辑
System.out.println("处理业务逻辑...");
int i = 3/0;//出现错误
//3. 手动签收
channel.basicAck(deliveryTag,true);
} catch (Exception e) {
//e.printStackTrace();
System.out.println("出现异常，拒绝接受");
//4.拒绝签收，不重回队列 requeue=false
channel.basicNack(deliveryTag,true,false);
}
}
}<rabbit:listener ref="dlxListener" queue-names="test_queue_dlx"> </rabbit:listener>
延迟队列，即消息进入队列后不会立即被消费，只有到达指定时间后，才会被消费。c
需求：
1.下单后，30分钟未支付，取消订单，回滚库存。
2.新用户注册成功7天后，发送短信问候。
实现方式：
定时器
延迟队列
定时器的实现方式不够优雅，我们采取延迟队列的方式
不过很可惜，在RabbitMQ中并未提供延迟队列功能。
但是可以使用：TTL+死信队列 组合实现延迟队列的效果。
配置
<!--
延迟队列：
1. 定义正常交换机（order_exchange）和队列(order_queue)
2. 定义死信交换机（order_exchange_dlx）和队列(order_queue_dlx)
3. 绑定，设置正常队列过期时间为30分钟
-->
<!-- 定义正常交换机（order_exchange）和队列(order_queue)-->
<rabbit:queue id="order_queue" name="order_queue">
<!-- 绑定，设置正常队列过期时间为30分钟-->
<rabbit:queue-arguments>
<entry key="x-dead-letter-exchange" value="order_exchange_dlx" />
<entry key="x-dead-letter-routing-key" value="dlx.order.cancel" />
<entry key="x-message-ttl" value="10000" value-type="java.lang.Integer"/>
</rabbit:queue-arguments>
</rabbit:queue>
<rabbit:topic-exchange name="order_exchange">
<rabbit:bindings>
<rabbit:binding pattern="order.#" queue="order_queue"></rabbit:binding>
</rabbit:bindings>
</rabbit:topic-exchange>
<!-- 定义死信交换机（order_exchange_dlx）和队列(order_queue_dlx)-->
<rabbit:queue id="order_queue_dlx" name="order_queue_dlx"></rabbit:queue>
<rabbit:topic-exchange name="order_exchange_dlx">
<rabbit:bindings>
<rabbit:binding pattern="dlx.order.#" queue="order_queue_dlx"></rabbit:binding>
</rabbit:bindings>
</rabbit:topic-exchange>生产端测试
@Test
public void testDelay() throws InterruptedException {
//1.发送订单消息。 将来是在订单系统中，下单成功后，发送消息
rabbitTemplate.convertAndSend("order_exchange","order.msg","订单信息：id=1,time=2019年8月17日16:41:47");
/*//2.打印倒计时10秒
for (int i = 10; i > 0 ; i--) {
System.out.println(i+"...");
Thread.sleep(1000);
}*/
}消费端监听
public class OrderListener implements ChannelAwareMessageListener {
@Override
public void onMessage(Message message, Channel channel) throws Exception {
long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
//1.接收转换消息
System.out.println(new String(message.getBody()));
//2. 处理业务逻辑
System.out.println("处理业务逻辑...");
System.out.println("根据订单id查询其状态...");
System.out.println("判断状态是否为支付成功");
System.out.println("取消订单，回滚库存....");
//3. 手动签收
channel.basicAck(deliveryTag,true);
} catch (Exception e) {
//e.printStackTrace();
System.out.println("出现异常，拒绝接受");
//4.拒绝签收，不重回队列 requeue=false
channel.basicNack(deliveryTag,true,false);
}
}
}<rabbit:listener ref="orderListener" queue-names="order_queue_dlx"> </rabbit:listener>
Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!
![[Web front-end] Démarrage rapide de Node.js](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
