How to use SpringBoot+RabbitMQ to achieve reliable message transmission

Environment configuration

SpringBootIntegrationRabbitMQrealizes the sending of messages.

1. Addmavendependencies

 org.springframework.boot spring-boot-starter   org.springframework.boot spring-boot-starter-web   org.springframework.boot spring-boot-starter-amqp 

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 messageshello, 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 toRabbitMQ

• Lost on the storage side:RabbitMQThe 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

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

Production phase

The production phase uses therequest confirmation mechanismto 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 fromproducertoswitch, there isconfirmCallbackConfirm mode. After the message is successfully sent, the message will call the methodconfirm(CorrelationData correlationData, boolean ack, String cause), and determine whether the message is successfully sent based onack.

Messages fromswitchtoqueuehavereturnCallbackreturn mode.

Send messageproduct messageThe 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 persistenceof. First, set the queue and switch correctly, and modify the queue for consumption monitoring so thatmessages 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 restartsRabbitMQ, 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 defaultackAutomatic 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 addingchannel.basicAck, restart the project

The queue message will be deleted

basicAckThe last parameter of the methodmultiplemeans deleting the previous queue.

multipleis set totrue, 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!