RbbitMQ(v) — 100% message delivery consumption

A: Introduction overview

The producer produces the message to the consumer, and the producer sends the message to the exchange, which routes the message to the queue store, and the consumer consumes the message. In the absence of any Settings, messages can be lost in the following ways:

1. The consumer sent the message to the switch because the message was lost for internal RabbitMQ reasons
2. The exchange routes the message to the queue because the queue does not exist and so on causing the message to be lost
3. Messages stored in the queue are lost when the RabbitMQ service is down before consumers consume them
4. When a consumer consumes a message, the consumer breaks down and the message is not processed, resulting in message loss

In view of the above situation, this paper will describe how to ensure the reliability of message delivery according to each node. At the same time, it may cause a series of problems such as message duplication under the guarantee of reliability, which is also the focus of this paper

Two: transaction TX

Understanding transactions in a database is to atomize multiple operations and commit rollback to achieve data consistency. RabbitMQ transactions can be different from previous transactions such as database transactions, and of course messages can be committed and rolled back. Transactions are one of the solutions to the problem of message loss when a producer sends a message to an exchange

2.1 Related Operations

The implementation of a transaction in RabbitMQ has two main steps:

1. Set the channel for communication to transaction mode
2. Transaction commit/transaction rollback

2.2 Code Examples

        // Set the channel to transaction mode
        channel.txSelect();
        // Send a message
        try {
            channel.basicPublish(bindingKey, bindingKey, false.false.null, messgae.getBytes("UTF-8"));
            // Commit the transaction
            channel.txCommit();
        }catch (Exception e){
            // The exception rolls back the transaction. When an exception occurs, you can try to resend or record the transaction
            channel.txRollback();
        }
Copy the code

2.3 summarize

The transaction mechanism has to wait for the last message to be sent before a second message can be sent, which would be a waste of RabbitMQ. If you want to send multiple messages, you can only cycle the operation, but note that if the channel is not set to transaction mode, the transaction operation cannot be performed, or an exception will be thrown. Finally, setting the channel to transaction mode only requires one operation

Three: Confirm

Transactions can be disastrous for RabbitMQ performance and offer a new lightweight way to handle producer-to-switch message loss. Confirm. There are a lot of places in this operation code that say wait for confirmation, batch confirmation, and asynchronous confirmation. Everything is for production, so this article will only cover asynchronous validation for production

3.1 Related Operations

The production send validation implementation for RabbitMQ consists of three parts:

• Set channel to confirm mode
• Add confirmation Listener
• Handling monitoring results

3.2 Code Examples

• Each channel sent to RabbitMQ’s Broker will have a unique encoding
• You are advised to use an ordered queue at the production end to store sent messages for easy deletion after confirmation
• When creating a Channel Channel, you can specify a unique code that identifies the Channel

        // Set the confirm message sending mechanism
        channel.confirmSelect();
        // Add confirmation mechanism listener
        channel.addConfirmListener(new ConfirmListener() {
            /** * successfully confirm * deliveryTag represents the unique identifier of the message * multiple Confirm whether this is a batch operation */
            @Override
            public void handleAck (long deliveryTag, boolean multiple) throws IOException {
                // Delete messages from the ordered collection
                if (! multiple){
                    // Delete messages based on coordinates
                }else {
                    // Delete messages in batches}}/** * failed to confirm * deliveryTag indicates the unique identifier of the message * multiple Specifies whether the operation is batch operation */
            @Override
            public void handleNack (long deliveryTag, boolean multiple) throws IOException {
                // Retries can be performed according to deliveryTag, etc}});Copy the code

3.3 Precautions

• Coexistence: Transactions and validation mechanisms cannot coexist, or an exception will occur
• Check: The Channels bar Mod attribute T stands for transaction and C stands for listening confirmation
• Ordered: Since RabbitMQ generates a sequence of deliveryTags that automatically increments from small to large, it is best to store messages in order to locate and manipulate them

Four: Mandatory

The exchange does not store messages, and all messages are routed to the queue store. If the intermediate process message is lost, it is an unknown error to the producer without setting it. The Mandatory implementation is similar to the Confirm implementation by adding a monitoring Listener. This parameter listener was described in detail in the previous article Messages and Queues

4.1 Related Operations

When a message arrives at the exchange but does not match the queue route store. If Mandatory is used to listen on processing, the following operations are required:

1. BasicPublish () is sent with the mandatory parameter set to true
2. Add MandatoryListener to channel

4.2 Code Examples

• First, you can see that messages cannot be routed to queues because the bindingKey and routingKey do not match
• You can then see that setting the Mandatory parameter to true when sending messages adds MANDATORY listening
• Finally, it can be seen that ReturnLisrtener listener is added on the channel to obtain the parameter information of unrouted message
  

  

4.3 Parameter Description

The ReturnListener contains only one method, handleReturn(), which contains a series of parameters, the meanings of which are shown in the following table:

Five: Standby switch

Mandatory added ReturnListener listening requires additional logic in the sending message code, which is not good news for feature specificity. RabbitMQ does not detect this logic at all, which is not conducive to subsequent code maintenance. Therefore, an alternate exchange is proposed, which is bound when the exchange is created and forwarded to the alternate exchange if the message from the exchange does not find the routing queue

5.1 Related Operations

The principle of a standby switch is similar to that of a switch bound to a switch. Note the following:

1. Use the Map parameter to bind the standby switch when creating the switch
2. The standby exchange receives routed messages without changing any properties, including the routingKey
3. The standby switch can be set up as a built-in switch

5.2 Code Examples

Bind the standby switch with the Map parameter to verify the effect. RoutingKey is set to the standby switch routingKey. You can look at the fifth parameter when the standby switch is created. It is also mentioned above that the optimal setting is the built-in switch, with the property internal

5.3 Test Results

Finally, a message is displayed in the standby switch, proving that the result is correct. The standby switch can be used as a solution when messages are not routed correctly

Six: queues and message persistence

This issue has been addressed in detail in the previous article on queues and queued messages. For the sake of complete message delivery reliability, the persistence of queues and queued messages is described here again. Note the following points:

Just as message persistence on a separate queue does not enable message persistence, so does message persistence on a separate queue. Both the queue and the queue message need to be persisted

6.1 Queue Persistence

Persistency The operation of writing queue information to disk for persistency and automatically restoring the data when the RabbitMQ service fails and restarts is called persistency. The implementation simply needs to set the persistence parameter to true when creating the queue, as shown below:

6.2 Queue Message Persistence

Restart the RabbitMQ service and find that the queue is restored, but the message data in the queue is not. Because queue message persistence needs to be set when messages are sent, it would not otherwise be written to disk for saving. The code looks like this:

• The send message method parameter list requires passing BasicProperties, which is designed using the Builder pattern, where deliveryMod represents message persistence. 1 The default value is not persisted. 2 The message is persisted and written to disk
• The MessageProperties class encapsulates the common family of BasicProperties objects and can be used directly

6.3 summarize

Queued persistence + queued message persistence = full persistence, persistence is a burden on RabbitMQ application performance, range data can be persisted according to data type. More important data such as order data, payment data and so on can be persisted to minimize message loss

Seven: consumer confirmation

The producer message has been delivered and routed to the queue storage. When the consumer consumes, the consumption application is down and the consumption logic is incomplete. The outage is also a key link to ensure that the message is delivered and consumed 100%. RabbitMQ provides a consumer confirmation mechanism for this purpose, with which the RabbitMQ application will delete messages only if and only if the consumer confirms

By default, the consumer will automatically confirm that messages are deleted when retrieved from the RabbitMQ application service, which is the cause of message loss. Therefore, the autoAck parameter needs to be set to false when consuming the message in order to implement the subsequent logic of manually controlling the message acknowledgement

7.1 basicAck

Message confirmation, with parameters including deliveryTag and multiple. Function consistent with producer confirmation:

• DeliveryTag: The RabbitMQ application generates a unique number for each message, and both the producer and consumer need to act on the message according to the code
• Multiple: Batch operation. That is, messages whose encoding is smaller than the encoding of this operation will be performed in the same operation

7.2 Message Rejection

There are two APIS for message rejection, basicReject() and basicNack(). The only difference between the two apis is that the former cannot batch manipulate multiple. Both contain the following two parameter attributes:

• DeliveryTag: The RabbitMQ application generates a unique number for each message, and both the producer and consumer need to act on the message according to the code
• Requeue: Whether to put back into the queue, where discarded messages with dead-letter forwarding are routed to the configured dead-letter exchange