When SpringBoot integrates Rabbitmq, we enable the manual confirmation mode of the message publisher and configure it as follows:
# Message middleware
spring:
rabbitmq:
host: 127.0.0.1
port: 5672
username: guest
password: guest
#Message producer confirmation mode
publisher-confirm-type: correlated
#Message producer callback is enabled
publisher-returns: true
After the above configuration is enabled, we need to manually write callbacks on the producer side as follows:
@Slf4j
@SpringBootTest
public class DHomeMessageApplicationTests {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void sendMessage(){
for (int i = 0; i < 3; i + + ) {
try {
Thread.sleep(2000);
SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
MessageDTO messageDTO = new MessageDTO();
messageDTO.setTitle("hello rabbitmq I am a handsome guy!!");
messageDTO.setContext(format.format(new Date()));
CorrelationData correlationData = new CorrelationData();
String messageStr = JSONObject.toJSONString(messageDTO);
ReturnedMessage returnedMessage = new ReturnedMessage(new Message(messageStr.getBytes()), 1, "1", MqConstants.EXCHANGE_NAME, "");
correlationData.setReturned(returnedMessage);
// MqConstants.EXCHANGE_NAME switch name
// MqConstants.QUEUE_NAME queue name
rabbitTemplate.convertAndSend(MqConstants.EXCHANGE_NAME,MqConstants.QUEUE_NAME,messageDTO,correlationData);
// Monitor whether the message is sent to the switch callback
rabbitTemplate.setConfirmCallback(((correlationData1, b, s) -> {
if(b){
log.error("Message sent successfully");
}else {
log.error("Failed to send message to switch");
ReturnedMessage returned = correlationData.getReturned();
Message message = returned.getMessage();
byte[] body = message.getBody();
String s1 = new String(body);
log.info("ConfirmCallback: " + "Related data:" + s1 );
log.info("ConfirmCallback: " + "Confirm whether the switch has been reached:" + b);
log.info("ConfirmCallback: " + "Reason:" + s);
}
}));
// The message was successfully sent to the switch, but failed to be sent to the queue. Callback
rabbitTemplate.setReturnsCallback(returnedMessage1 -> {
String s = returnedMessage1.getMessage().getBody().toString();
log.error("Failed to send message to queue");
log.error("Message content: {}",s);
});
}catch (Exception e){
log.error("Failed to send message");
e.printStackTrace();
}
}
}
}
When the above code is executed, an error will occur:
The reason for the above error is that rabbitMqTemplate is a singleton by default, and a rabbitMqTemplate can only listen to one ConfirCallback. The online solution is:
Modify rabbitMqTemplate to multi-instance mode: as follows: We define a rabbitMqTemplate in multi-instance mode and inject custom multi-instance Bean when using it
@Bean(name = "rabbitTemplatePrototype")
//Set to multiple instances, use different beans for each injection
@Scope("prototype")
public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory){
RabbitTemplate rabbitTemplate = new RabbitTemplate();
rabbitTemplate.setConnectionFactory(connectionFactory);
return rabbitTemplate;
}
In the producer use:
@Resource(name = "rabbitTemplatePrototype")
private RabbitTemplate rabbitTemplate;
Run again: After using our customized multi-case template, running the test code will still report the following error:
At this time, the article on the Internet said that calling in the Controller and specifying this Controller also has multiple instances, we made the modification:
The following are the execution results:
Still getting an error. . . . . .
I don’t care about the error for now, but it would be too low if the message sending code is written in the controller (violating the three-layer design).
Here are some possible solutions:
Directly in the Bean that defines the rabbitMqTemplate template, set the callback method as follows:
@Bean(name = "rabbitTemplate")
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){<!-- -->
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
//Set the callback for successful messages sent from the producer to the rabbitmq broker (to ensure that the information reaches the broker)
rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {<!-- -->
// ack=true: The message was successfully sent to Exchange
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {<!-- -->
ReturnedMessage returned = correlationData.getReturned();
Message message = returned.getMessage();
byte[] body = message.getBody();
String s = new String(body);
boolean valid = JSONObject.isValid(s);
MessageDTO messageDTO=null;
if(valid){<!-- -->
messageDTO = JSONObject.parseObject(s, MessageDTO.class);
}
log.info("ConfirmCallback: " + "Related data:" + messageDTO );
log.info("ConfirmCallback: " + "Confirm whether the switch has been reached: " + ack);
log.info("ConfirmCallback: " + "Cause:" + cause);
}
});
//Set the callback for failure to send information from the switch to the queue
rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {<!-- -->
@Override
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {<!-- -->
log.info("ReturnCallback: " + "Message: " + message);
log.info("ReturnCallback: " + "Response code: " + replyCode);
log.info("ReturnCallback: " + "Response information:" + replyText);
log.info("ReturnCallback: " + "Exchange:" + exchange);
log.info("ReturnCallback: " + "Routing Key:" + routingKey);
}
});
// When it is true, if the message cannot be matched to the queue through the exchange, it will be returned to the producer. When it is false, if it cannot be matched, it will be discarded directly.
rabbitTemplate.setMandatory(true);
//Set the conversion when sending serialization
rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
return rabbitTemplate;
}
To modify the producer code, you only need to send a message: as follows
The running results are as follows: The problem of Only one ConfirmCallback is supported by each RabbitTemplate has not occurred again.
The above is a demonstration. The switch does not report an error when the message is sent to the switch. Below we simulate that the message is sent to the switch successfully but fails when sent to the queue.
After the consumer obtains the message, it confirms that it has been sent to the queue.