How to use SpringBoot+RabbitMQ to achieve reliable message transmission

Environment configuration

SpringBoot Integration RabbitMQ realizes the sending of messages.

1. Add maven dependencies

<dependency>
     <groupId>org.springframework.boot</groupId>
     <artifactId>spring-boot-starter</artifactId>
 </dependency>

 <dependency>
     <groupId>org.springframework.boot</groupId>
     <artifactId>spring-boot-starter-web</artifactId>
 </dependency>

 <dependency>
     <groupId>org.springframework.boot</groupId>
     <artifactId>spring-boot-starter-amqp</artifactId>
 </dependency>

2. Add application.yml configuration file

spring:
  rabbitmq:
    host: 192.168.3.19
    port: 5672
    username: admin
    password: xxxx

3. Configure switches, queues and bindings

@Bean
public DirectExchange myExchange() {
    DirectExchange directExchange = new DirectExchange("myExchange");
    return directExchange;
}

@Bean
public Queue myQueue() {
    Queue queue = new Queue("myQueue");
    return queue;
}

@Bean
public Binding binding() {
    return BindingBuilder.bind(myQueue()).to(myExchange()).with("myRoutingKey");
}

4. Production sends messages

@Autowired
private RabbitTemplate rabbitTemplate;

@GetMapping("/send")
public String send(String message) {
    rabbitTemplate.convertAndSend("myExchange","myRoutingKey",message);
    System.out.println("【发送消息】" + message)
    return "【send message】" + message;
}

5. Consumers receive messages

@RabbitListener(queuesToDeclare = @Queue("myQueue"))
public void process(String msg, Channel channel, Message message) {
    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    Date date = new Date();
    String time = sdf.format(date);
    System.out.println("【接收信息】" + msg + " 当前时间" + time);

6. Call the production end to send messages hello, console output:

[Send message] hello
[Receive message] hello Current time 2022-05-12 10:21:14

Indicates that the message has been successfully received.

Message loss analysis

From production to consumption of a message, message loss may occur in the following stages:

• Lost on the production side: The producer cannot transmit to RabbitMQ

• Lost on the storage side: RabbitMQ The storage itself is down

• Lost at the consumer end: The storage cannot be sent to the consumer end due to network problems, or consumption hangs and normal consumption cannot be sent

RabbitMQ Provides good support for reliable transmission from the production side, storage side, and consumer side.

Production phase

The production phase uses the request confirmation mechanism to ensure reliable transmission of messages. After sending a message to the RabbitMQ server, RabbitMQ receives the message and returns a request confirmation to the sender, indicating that the RabbitMQ server has successfully received the message.

Configuration application.yml

spring:
  rabbitmq:
    # 消息确认机制 生产者 -> 交换机
    publisher-confirms: true
    # 消息返回机制  交换机 -> 队列
    publisher-returns: true

Configuration

@Configuration
@Slf4j
public class RabbitConfig {

    @Autowired
    private ConnectionFactory connectionFactory;

    @Bean
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                log.info("【correlationData】:" + correlationData);
                log.info("【ack】" + ack);
                log.info("【cause】" + cause);
                if (ack) {
                    log.info("【发送成功】");
                } else {
                    log.info("【发送失败】correlationData:" + correlationData + " cause:" + cause);
                }
            }
        });
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                log.warn("【消息发送失败】");
                log.info("【message】" + message);
                log.info("【replyCode】" + replyCode);
            }
        });

        return rabbitTemplate;
    }
}

Messages from producer to switch, there is confirmCallback Confirm mode. After the message is successfully sent, the message will call the method confirm(CorrelationData correlationData, boolean ack, String cause), and determine whether the message is successfully sent based on ack.

Messages from switch to queue have returnCallback return mode.

Send message product message The console output is as follows:

[Send message]product message
[Receive message]product message Current time 2022-05 -12 11:27:56
[correlationData]:null
[ack]true
[cause]null
[Send successfully]

The production end simulation message is lost

There are two solutions here:

• Close the broker immediately after sending the message. The latter shuts down the network, but after the broker is closed, the console will always report an error and send the message. Also reported a 500 error.

• Send non-existent switch:

// myExchange 修改成 myExchangexxxxx
rabbitTemplate.convertAndSend("myExchangexxxxx","myRoutingKey",message);

Result:

[correlationData]:null
【ack】false
【cause】channel error; protocol method: #method(reply-code=404, reply-text=NOT_FOUND - no exchange 'myExchangexxxxx' in vhost '/', class-id =60, method-id=40)
[Failed to send]

When sending fails, you can retry the message

The switch is correct and the queue that does not exist is sent:

The switch receives the message and returns a success notification. The console output:

[correlationData]:CorrelationData [id=7d468b47-b422-4523-b2a2-06b14aef073c]
[ack ]true
[cause]null
[Send successfully]

The switch did not find the queue and returned failure information:

[Message sending failed]
[message]product message
[replyCode]312

RabbitMQ

Enable queue persistence, create queues and switchesThe default configuration is persistence of. First, set the queue and switch correctly, and modify the queue for consumption monitoring so that messages are stored in the queue.

Modify the persistence of the queue to non-persistence:

@Bean
public Queue myQueue() {
    Queue queue = new Queue("myQueue",false);
    return queue;
}

After sending the message, the message is stored in the queue, and then restarts RabbitMQ, the message no longer exists.
Set queue persistence:

@Bean
public Queue myQueue() {
    Queue queue = new Queue("myQueue",true);
    return queue;
}

After restarting, the messages in the queue still exist.

Consumer end

The consumer end starts by default ack Automatic confirmation mode. When the queue message is received by the consumer, it will be automatically deleted from the queue regardless of whether there is a message from the consumer end. news. Therefore, in order to ensure that the consumer can successfully consume the message, change the automatic mode to manual confirmation mode:

Modify the application.yml file

spring:
  rabbitmq:
    # 手动消息确认
    listener:
      simple:
        acknowledge-mode: manual

After consuming and receiving the message, manual confirmation is required:

channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);

@RabbitListener(queuesToDeclare = @Queue("myQueue"))
public void process(String msg, Channel channel, Message message) {
    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    Date date = new Date();
    String time = sdf.format(date);
    System.out.println("【接收信息】" + msg + " 当前时间" + time);
    System.out.println(message.getMessageProperties().getDeliveryTag());
    try {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    } catch (IOException e) {
        e.printStackTrace();
    }

}

If not added:

channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);

Send two messages

After the message is received, there is no confirmation and it is put back into the queue:

Restart the project. After that, the messages in the queue will be sent to the consumer, but without ack confirmation, they will continue to be put back into the queue.

After adding channel.basicAck, restart the project

The queue message will be deleted

basicAck The last parameter of the method multiple means deleting the previous queue.

multiple is set to true, and all subsequent queues are cleared

The above is the detailed content of How to use SpringBoot+RabbitMQ to achieve reliable message transmission. For more information, please follow other related articles on the PHP Chinese website!