Méthodes d'application avancées RabbitMQ en Java

1. Livraison fiable des messages

 Lors de l'utilisation de RabbitMQ, si le producteur souhaite savoir si le message est correctement transmis au commutateur et à la file d'attente correspondants lors de la transmission du message, il existe deux méthodes. peut être utilisé pour contrôler le mode de fiabilité de la livraison des messages. RabbitMQ 的时候，生产者在进行消息投递的时候如果想知道消息是否成功的投递到对应的交换机和队列中，有两种方式可以用来控制消息投递的可靠性模式 。

 由上图的整个消息的投递过程来看，生产者的消息进入到中间件中会首先到达交换机，然后再从交换机传递到队列中去，也就是分为两步走战略。那么消息的丢失情况也就是会出现在这两个阶段中，RabbitMQ 贴心的为我们提供了针对于这两个部分的可靠新传递模式：

• confirm 模式。

• return 模式。

 利用这两个回调模式来确保消息的传递可靠。

 1.1、确认模式

 消息从生产者到交换机之间传递会返回一个 confirmCallback 的回调。可以直接在 rabbitTemplate 实例中进行确认逻辑的设置。如果是使用 XML 配置的话需要在工厂配置开启 publisher-confirms="true"，YAML 的配置就直接 publisher-confirm-type: correlated，他默认是 NONE ，需要手动开启。

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq.xml")
public class Producer {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void producer() throws InterruptedException {
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {
                System.out.println();
                if (!b) {
                    //	消息重发之类的处理
                    System.out.println(s);
                } else {
                    System.out.println("交换机成功接收消息");
                }
            }
        });
        rabbitTemplate.convertAndSend("default_exchange", "default_queue",
                "hello world & beordie");
        TimeUnit.SECONDS.sleep(5);
    }
}

 上面的确认是由一个 confirm 的函数执行的，里面携带了三个参数，第一个是配置的相关信息，第二个表示交换机是否成功的接收到消息，第三个参数是指没有成功接收消息的原因。

 1.2、退回模式

 从交换机到消息队列投递失败会返回一个 returnCallback 。在工厂配置中开启回退模式 publisher-returns="true" ，设置交换机处理消息失败的模式（默认 false 直接将消息进行丢弃），添加退回处理的逻辑。

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq.xml")
public class Producer {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void producer() throws InterruptedException {
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                //  重发逻辑处理
                System.out.println(message.getBody() + " 投递消息队列失败");
            }
        });
        rabbitTemplate.convertAndSend("default_exchange", "default_queue",
                "hello world & beordie");
        TimeUnit.SECONDS.sleep(5);
    }
}

 returnedMessage 中携带五个参数、分别指的是消息对象、错误码、错误信息、交换机、路由键。

 1.3、确认机制

 在消费者抓取消息队列中的数据取消费之后会有一个确认机制进行消息的确认，防止因为抓取消息之后但没有消费成功而导致的消息丢失。有三种确认方式：

• 自动确认：acknowledge="none"

• 手动确认：acknowledge="manual"

• 根据异常情况确认：acknowledge="auto"

 其中自动确认是指一旦消息被消费者抓取就自动默认成功，并将消息从消息队列中进行移除，如果这个时候消费端消费出现问题，那么也会是默认消息消费成功，但是实际上是没有消费成功的，也就是当前的消息丢失了。默认的情况就是自动确认机制。

 如果设置手动确认的方式，就需要在正常消费消息之后进行回调确认 channel.basicAck()，手动签收。如果业务处理过程中发生了异常则调用 channel.basicNack() 重新发送消息。

 首先需要在队列绑定时进行确认机制的配置，设置为手动签收。

<!-- 绑定队列 -->
<rabbit:listener-container connection-factory="rabbitFactory" auto-declare="true" acknowledge="manual">
    <rabbit:listener ref="rabbirConsumer" queue-names="default_queue"/>
</rabbit:listener-container>

 生产者一端不用更改，只需要改变消费者的实现进行消息自动签收就可以了，正常执行业务则签收消息，业务发生错误则选择消息拒签，消息重发或者丢弃。

public class ConsumerAck implements ChannelAwareMessageListener {
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        //  消息唯一ID
        long tag = message.getMessageProperties().getDeliveryTag();
        try {
            String msg = new String(message.getBody(), "utf-8");
            channel.basicAck(tag, true);
            System.out.println("接收消息: " + msg);
        } catch (Exception e) {
            System.out.println("接收消息异常");
            channel.basicNack(tag, true, true);
            e.printStackTrace();
        }
    }
}

 里面涉及三个简单的签收函数，一是正确签收的 basicAck ，二是单条拒签的 basicReject ，三是批量拒签的 basicNack

•  De l'image ci-dessus Du point de vue de l'ensemble du processus de livraison des messages, le message du producteur entrant dans le middleware arrivera d'abord au commutateur, puis sera transmis du commutateur à la file d'attente, ce qui est une stratégie en deux étapes. Ensuite, la perte de message se produira au cours de ces deux étapes. RabbitMQ nous fournit judicieusement un nouveau mode de livraison fiable pour ces deux parties :

mode de confirmation
• .

mode retour
• .

 Utilisez ces deux modes de rappel pour garantir une livraison fiable des messages.

1.1. Mode de confirmation

🎜  Lorsqu'un message est transmis du producteur au commutateur, un rappel confirmCallback sera renvoyé. La logique de confirmation peut être définie directement dans l'instance rabbitTemplate. Si vous utilisez XML pour configurer, vous devez activer 🎜publisher-confirms="true"🎜 dans la configuration d'usine. La configuration de YAML est directement 🎜publisher-confirm-. tapez : corrélé🎜. La valeur par défaut est NONE et doit être activée manuellement. 🎜

<!-- 绑定队列 -->
<rabbit:listener-container connection-factory="rabbitFactory" auto-declare="true"
                           acknowledge="manual" prefetch="1">
    <rabbit:listener ref="rabbirConsumer" queue-names="default_queue"/>
</rabbit:listener-container>

🎜 La confirmation ci-dessus est exécutée par une fonction confirm, qui comporte trois paramètres. Le premier concerne les informations relatives à la configuration et le second indique si le commutateur l'a reçu avec succès. , le troisième paramètre fait référence à la raison pour laquelle le message n'a pas été reçu avec succès. 🎜

1.2. Mode de retour

🎜 Si la livraison du commutateur vers la file d'attente des messages échoue, un returnCallback sera renvoyé. Activez le mode de secours 🎜publisher-returns="true"🎜 dans la configuration d'usine, définissez le mode pour que le commutateur échoue dans le traitement des messages (la valeur par défaut est false et le message sera directement rejeté) et ajoutez une logique de traitement de secours. 🎜

<rabbit:queue id="default_queue" name="default_queue" auto-declare="true">
    <rabbit:queue-arguments>
        <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer"/>
    </rabbit:queue-arguments>
</rabbit:queue>

🎜 returnedMessage comporte cinq paramètres, qui font respectivement référence à l'objet du message, au code d'erreur, au message d'erreur, au commutateur et à la clé de routage. 🎜

1.3. Mécanisme de confirmation

🎜 Une fois que le consommateur a récupéré les données dans la file d'attente des messages et annulé la consommation, il y aura un mécanisme de confirmation pour confirmer le message afin d'éviter l'échec de la consommation après avoir récupéré le message. .les messages sont perdus. Il existe trois méthodes de confirmation : 🎜🎜🎜🎜🎜Confirmation automatique🎜 : acknowledge="none"🎜🎜🎜🎜🎜Confirmation manuelle🎜 : acknowledge="manual"🎜🎜 🎜 🎜🎜Confirmer en fonction de conditions anormales🎜 : acknowledge="auto"🎜🎜🎜  La confirmation automatique signifie qu'une fois le message capturé par le consommateur, il sera automatiquement par défaut succès et le message sera supprimé de Supprimez-le de la file d'attente des messages. S'il y a un problème avec la consommation du consommateur à ce moment-là, la consommation du message par défaut sera réussie, mais en fait la consommation n'est pas réussie, c'est-à-dire le message actuel. est perdu. La valeur par défaut est le mécanisme de confirmation automatique. 🎜🎜 Si vous définissez la méthode de confirmation manuelle, vous devez effectuer une confirmation de rappel channel.basicAck() après la consommation normale des messages et vous y connecter manuellement. Si une exception se produit pendant le traitement métier, appelez channel.basicNack() pour renvoyer le message. 🎜🎜 Tout d'abord, vous devez configurer le mécanisme de confirmation lors de la liaison de la file d'attente et le définir sur la signature manuelle. 🎜

MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
    @Override
    public Message postProcessMessage(Message message) throws
        AmqpException {
        //	设置 message 的过期时间
        message.getMessageProperties().setExpiration("5000");
        //	返回该消息
        return message;
    }
};
rabbitTemplate.convertAndSend("exchange", "route", "msg", messagePostProcessor);

🎜  Il n'est pas nécessaire de changer le côté producteur. Il vous suffit de modifier l'implémentation du consommateur pour signer automatiquement le message. Si l'entreprise est exécutée normalement, le message sera signé. le message sera rejeté et le message sera renvoyé ou rejeté. 🎜

<rabbit:queue id="default_queue" name="default_queue" auto-declare="true">
    <rabbit:queue-arguments>
        <!-- 死信交换机 -->
        <entry key="x-dead-letter-exchange" value-type="dlx_exchane"/>
        <!-- 路由 -->
        <entry key="x-dead-letter-routing-key" value-type="dlx_routing"/>
        <!-- 队列过期时间 -->
        <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer"/>
        <!-- 队列长度 -->
        <entry key="x-max-length" value-type="java.lang.Integer" value="10"/>
    </rabbit:queue-arguments>
</rabbit:queue>

🎜 Cela implique trois fonctions de signature simples, l'une est basicAck pour une signature correcte, la seconde est basicReject pour un rejet unique et la troisième est pour un lot. rejet basicNack. 🎜🎜🎜🎜🎜basicAck🎜 Le premier paramètre indique l'ID unique du message dans le canal, qui concerne uniquement le canal actuel ; le deuxième paramètre indique s'il faut accepter par lots. S'il est faux, il acceptera uniquement de signer. pour un message avec l'ID actuel, il est supprimé de la file d'attente des messages, et si c'est vrai, les messages précédant cet ID seront signés ensemble. 🎜🎜🎜🎜🎜basicReject🎜 Le premier paramètre indique toujours l'ID unique du message, et le deuxième paramètre indique s'il faut remettre en file d'attente pour l'envoi false indique que le message est rejeté directement ou qu'il existe une file d'attente de lettres mortes pour le recevoir true. indique que le message est remis en file d'attente. Pour envoyer des messages, toutes les opérations concernent uniquement le message en cours. 🎜🎜🎜🎜🎜basicNack🎜 a un paramètre de plus que le second, qui est une valeur booléenne au milieu, indiquant s'il faut regrouper. 🎜

2、消费端限流

 在用户请求和DB服务处理之间增加消息中间件的隔离，使得突发流量全部让消息队列来抗，降低服务端被冲垮的可能性。让所有的请求都往队列中存，消费端只需要匀速的取出消息进行消费，这样就能保证运行效率，也不会因为后台的阻塞而导致客户端得不到正常的响应（当然指的是一些不需要同步回显的任务）。

 只需要在消费者绑定消息队列时指定取出消息的速率即可，需要使用手动签收的方式，每进行一次的签收才会从队列中再取出下一条数据。

<!-- 绑定队列 -->
<rabbit:listener-container connection-factory="rabbitFactory" auto-declare="true"
                           acknowledge="manual" prefetch="1">
    <rabbit:listener ref="rabbirConsumer" queue-names="default_queue"/>
</rabbit:listener-container>

3、消息过期时间

 消息队列提供了存储在队列中消息的过期时间，分为两个方向的实现，一个是针对于整个队列中的所有消息，也就是队列的过期时间，另一个是针对当前消息的过期时间，也就是针对于单条消息单独设置。

 队列的过期时间设置很简单，只需要在创建队列时进行过期时间的指定即可，也可以通过控制台直接创建指定过期时间。一旦队列过期时间到了，队列中还未被消费的消息都将过期，进行队列的过期处理。

<rabbit:queue id="default_queue" name="default_queue" auto-declare="true">
    <rabbit:queue-arguments>
        <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer"/>
    </rabbit:queue-arguments>
</rabbit:queue>

 单条消息的过期时间需要在发送的时候进行单独的指定，发送的时候指定配置的额外信息，配置的编写由配置类完成。

 如果一条消息的过期时间到了，但是他此时处于队列的中间，那么他将不会被处理，只有当之后处理到时候才会进行判断是否过期。

MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
    @Override
    public Message postProcessMessage(Message message) throws
        AmqpException {
        //	设置 message 的过期时间
        message.getMessageProperties().setExpiration("5000");
        //	返回该消息
        return message;
    }
};
rabbitTemplate.convertAndSend("exchange", "route", "msg", messagePostProcessor);

 如果说同时设置了消息的过期时间和队列的过期时间，那么最终的过期时间由最短的时间进行决定，也就是说如果当前消息的过期时间没到，但是整个队列的过期时间到了，那么队列中的所有消息也自然就过期了，执行过期的处理策略。

4、死信队列

 4.1、死信概念

死信队列指的是死信交换机，当一条消息成为死信之后可以重新发送到另一个交换机进行处理，而进行处理的这个交换机就叫做死信交换机。

• 消息成为死信消息有几种情况

  队列的消息长度达到限制

  消费者拒接消息的时候不把消息重新放入队列中

  队列存在消息过期设置，消息超时未被消费

  消息存在过期时间，在投递给消费者时发现过期

 在创建队列时可以在配置中指定相关的信息，例如死信交换机、队列长度等等，之后的一系列工作就不由程序员进行操作了，MQ 会自己完成配置过的事件响应。

<rabbit:queue id="default_queue" name="default_queue" auto-declare="true">
    <rabbit:queue-arguments>
        <!-- 死信交换机 -->
        <entry key="x-dead-letter-exchange" value-type="dlx_exchane"/>
        <!-- 路由 -->
        <entry key="x-dead-letter-routing-key" value-type="dlx_routing"/>
        <!-- 队列过期时间 -->
        <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer"/>
        <!-- 队列长度 -->
        <entry key="x-max-length" value-type="java.lang.Integer" value="10"/>
    </rabbit:queue-arguments>
</rabbit:queue>

 4.2、延迟队列

 延迟队列指的是消息在进入队列后不会立即被消费，只有到达指定时间之后才会被消费，也就是需要有一个时间的判断条件。

 消息队列实际上是没有提供对延迟队列的实现的，但是可以通过 TTL + 死信队列 的方式完成，设置一个队列，不被任何的消费者所消费，所有的消息进入都会被保存在里面，设置队列的过期时间，一旦队列过期将所有的消息过渡到绑定的死信队列中。

 再由具体的消费者来消费死信队列中的消息，这样就实现了延迟队列的功能。

 例如实现一个下单超时支付取消订单的功能：

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!