Beispielanalyse für erweiterte Funktionen von Java RabbitMQ

Zuverlässige Zustellung von Nachrichten

Bei Verwendung von RabbitMQ als Absender der Nachricht hoffen wir, Nachrichtenverluste oder Zustellungsfehler zu verhindern. RabbitMQ bietet uns zwei Möglichkeiten, den Zustellzuverlässigkeitsmodus von Nachrichten zu steuern.

• Confirm-Bestätigungsmodus

• Return-Return-Modus

rabbitmq Der gesamte Nachrichtenübermittlungspfad ist:

producer—>rabbitmq-Broker—>exchange—>queue—& gt;Verbraucher

• Wenn die Nachricht vom Produzenten an die Börse gesendet wird, wird ein Bestätigungsrückruf zurückgegeben die zuverlässige Zustellung der Nachricht

Bestätigungsmodus

• Die Nachricht gibt einen Bestätigungsrückruf vom Produzenten an die Börse zurück

Nehmen Sie die Spring-Integration Rabbitmq als Beispiel, ändern Sie die Rabbitmq-Konfigurationsdatei, fügen Sie das Attribut „publisher-confirms“ in „connectionFactory“ hinzu und setzen Sie den Wert auf „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"/>

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

Da die Nachricht normalerweise an die Warteschlange gesendet wird, ist der zurückgegebene Ursachenwert leer. Wenn eine Ausnahme auftritt, ist die Ursache der Grund für die Ausnahme.

Rückgabemodus

Wenn die Nachricht nicht zugestellt werden kann In der Exchange-Warteschlange wird ein ReturnCallback zurückgegeben

Eine Nachricht wird nur zurückgegeben, wenn ein Fehler auftritt. Fügen Sie daher manuell einen Fehler hinzu und fügen Sie der gesendeten Nachricht den Routing-Wert return123 hinzu. Tatsächlich gibt es keine solche Route und die durch Ausführen zurückgegebene Nachricht lautet wie folgt.

Consumer Ack

ack bezieht sich auf Acknowledge, Bestätigung. Gibt die Bestätigungsmethode an, nachdem der Verbraucher die Nachricht erhalten hat. 🔜 (auf diese Weise ist die Verwendung umständlich, kein Lernen)

Die automatische Bestätigung bedeutet, dass der Verbraucher, sobald die Nachricht eintrifft, den Empfang automatisch bestätigt und die entsprechende Nachricht aus dem Nachrichtencache von RabbitMQ entfernt. Bei der tatsächlichen Geschäftsverarbeitung ist es jedoch sehr wahrscheinlich, dass die Nachricht verloren geht, wenn eine Nachricht empfangen wird und bei der Geschäftsverarbeitung eine Ausnahme auftritt. Wenn die manuelle Bestätigungsmethode festgelegt ist, müssen Sie nach erfolgreicher Geschäftsverarbeitung Channel.basicAck() aufrufen und sich manuell anmelden. Wenn eine Ausnahme auftritt, rufen Sie die Methode Channel.basicNack() auf, damit die Nachricht automatisch erneut gesendet werden kann.

Nehmen wir als Beispiel die Rabbitmq-Integration im Frühling. Die Bestätigungsmethode ist in der Rabbitmq-Konfigurationsdatei festgelegt. Der Überwachungsklassencode lautet wie folgt: Da eine Ausnahme auftritt, wird die Methode channel.basicNack () aufgerufen damit die Nachricht automatisch erneut gesendet werden kann, sodass die Endlosschleife den Inhalt ausgibt wird, und das wird passieren: Eine große Menge an Nachrichten werden alle sofort übertragen, aber unser einzelner Client kann nicht so viele Daten gleichzeitig verarbeiten. Wenn die Datenmenge extrem groß ist, ist es für uns definitiv unwissenschaftlich, sie zu begrenzen Fluss des Produktionsendes, da die Parallelität manchmal extrem groß ist. Und es ist sehr selten, dass wir das Verhalten des Benutzers einschränken können. Daher sollten wir den Fluss auf der Verbraucherseite begrenzen. Rabbitmq bietet eine QOS-Funktion (Quality of Service), d oder Consumer) werden vorher nicht bestätigt, verbrauchen keine neuen Nachrichten.

1. Stellen Sie sicher, dass der Bestätigungsmechanismus eine manuelle Bestätigung ist

2. Das Listener-Container-Konfigurationsattribut „perfetch = 1“ bedeutet, dass der Verbraucher jedes Mal eine Nachricht von MQ abruft und diese nicht weiter abruft, bis der Verbrauch erfolgt Manuell bestätigt.

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

Der Produzent sendet fünf Nachrichten

    @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...");
    }

Der Produzent kommentiert Channel.basicAck(deliveryTag,true) aus und bestätigt den Empfang der Nachricht nicht

<rabbit:listener-container connection-factory="connectionFactory"
acknowledge="manual">
.....

Nach dem Starten des Verbrauchers und dem erneuten Ausführen des Produzenten wird festgestellt, dass der Verbraucher hat fünf Nachrichten gesendet, tatsächlich hat der Produzent nur eine Nachricht erhalten, was den aktuellen begrenzenden Effekt erreicht hat

Bei der Beobachtung der RabbitMQ-Konsole wurde festgestellt, dass es 1 Unack-Nachricht gab. 4 Fertignachrichten haben den Verbraucher noch nicht erreicht. Dies steht im Einklang mit der aktuellen Grenzsituation von prefetchCount=1, die wir festgelegt haben.

• 

  把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>

Das obige ist der detaillierte Inhalt vonBeispielanalyse für erweiterte Funktionen von Java RabbitMQ. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!