Analyse d'exemples de fonctionnalités avancées Java RabbitMQ

Livraison fiable des messages

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

# 🎜🎜 # RabbitMQ L'ensemble du chemin de livraison des messages est: # 🎜🎜 ## 🎜🎜 # Producer & mdash; & gt; Rabbitmq Broker & mdash; & gt; Exchange & mdash; & gt; queue & mdash; & gt; consommateur # 🎜🎜 ## 🎜🎜 ## 🎜🎜 # # # # #

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

• Le message sera renvoyé par le producteur pour échanger confirmCallback
  Prenons l'exemple de l'intégration de Spring Rabbitmq, modifiez le fichier de configuration Rabbitmq, ajoutez l'attribut Publisher-Confirms dans connectionFactory et définissez la valeur sur true

<!--
* 确认模式：
* 步骤：
* 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"/>

rrree

• #🎜🎜 #

  É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

Mode retour

Le message provient de Si l'échange–>la livraison dans la file d'attente échoue, un returnCallback sera renvoyé

1 . Activez le mode de secours : editor-returns="true"

/*
 * 确认模式：
 * 步骤：
 * 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....");
    }

2 Configurez Exchange pour qu'il traite le message en cas d'échec. Mode : setMandatory, puis définissez ReturnCallBack

    <!-- 定义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"/>

. Le message ne sera renvoyé que si une erreur se produit, ajoutez donc manuellement une erreur et ajoutez la valeur de routage return123 au message envoyé. En fait, une telle route n'existe pas. Run Le message de retour est le suivant.

Consumer Ack

ack fait référence à l'accusé de réception, à la confirmation. Indique la méthode de confirmation une fois que le consommateur a reçu le message.

Il existe trois méthodes de confirmation :

Confirmation automatique : accusé de réception="aucun"

# 🎜🎜#Confirmation manuelle : accusé de réception="manuel"

Confirmer selon des conditions anormales : reconnaître="auto", (cette méthode est gênante à utiliser et ne learn)# 🎜🎜#

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.

Prenons l'exemple de l'intégration de Rabbitmq par Spring. Définissez la méthode de confirmation dans le fichier de configuration de Rabbitmq

•     @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">
  .....

  #. 🎜🎜#En raison d'une exception, appelez la méthode channel.basicNack(), lui permettant de renvoyer automatiquement le message, donc le contenu de sortie est une boucle infinie
Limite actuelle du consommateur
• #🎜🎜 #

Lorsque notre serveur Rabbitmq a un retard de dizaines de milliers de messages non traités et que nous ouvrons un client consommateur à volonté, cela se produira : Une énorme quantité de messages est tous envoyés instantanément. Poussés, mais notre seul client ne peut pas traiter autant de données en même temps. Lorsque la quantité de données est extrêmement importante, il n'est absolument pas scientifique pour nous de limiter le flux de production ! , parce que parfois la concurrence est extrêmement grande, et parfois la concurrence est extrêmement moindre, nous ne pouvons pas contraindre la fin de la production, c'est le comportement de l'utilisateur. Par conséquent, nous devrions limiter le flux du côté du consommateur. Rabbitmq fournit une fonction qos (qualité de service), c'est-à-dire sur la base d'une confirmation non automatique des messages, si un certain nombre de messages (la valeur Qos est définie pour le canal). ou consommateur) ne sont pas confirmés auparavant, ne consommez pas de nouveaux messages.

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);
        }
    }
}

#🎜🎜 #Après avoir démarré le consommateur et exécuté le producteur à ce moment-là, il a été constaté que le consommateur avait envoyé cinq messages, mais en fait le producteur n'a reçu qu'un seul message, atteignant l'effet limitant actuel

Observez la console lapinmq et trouvez un message de décompression. 4 messages prêts ne sont pas encore parvenus au consommateur. Cela est cohérent avec la situation limite actuelle de prefetchCount=1 que nous avons définie.

把channel.basicAck(deliveryTag,true)的注释取消掉，即可以自动确认收到消息，重新运行消费者，接收到了另外的四条消息

TTL(Time To Live)

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!