背景

这里后补直接上代码

最佳实践

主要从两个方面保证消息不丢失

RabbitMQ方面

1. 创建队列时开启持久化
2. 创建交换器时开启持久化
3. 创建镜像队列（可选）
4. 开启延迟队列（可选）

代码层面

1. 开启生产者到交换器回调参数
2. 开启交换器到队列回调参数
3. 开启消费者手动ack
4. 注意消费端打印日志，考虑手动补偿（可选）

实现代码

RabbitMQ配置文件

package com.jndj.core.config;import com.rabbitmq.client.ConnectionFactory;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;@Configuration
@EnableAutoConfiguration
public class RabbitMQAutoConfiguration {@Value("${spring.rabbitmq.host}")public String rmHost;@Value("${spring.rabbitmq.port}")public int rmPort;@Value("${spring.rabbitmq.username}")public String rmUsername;@Value("${spring.rabbitmq.password}")public String rmPassword;@Value("${spring.rabbitmq.virtual-host}")public String virtualHost;@Beanpublic CachingConnectionFactory connectionFactory() throws NoSuchAlgorithmException, KeyManagementException {ConnectionFactory rabbitConnectionFactory = new ConnectionFactory();rabbitConnectionFactory.setHost(rmHost);rabbitConnectionFactory.setPort(rmPort);rabbitConnectionFactory.setUsername(rmUsername);rabbitConnectionFactory.setPassword(rmPassword);rabbitConnectionFactory.setVirtualHost(virtualHost);rabbitConnectionFactory.useSslProtocol();CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(rabbitConnectionFactory);cachingConnectionFactory.setPublisherReturns(true);cachingConnectionFactory.setPublisherConfirms(true);return cachingConnectionFactory;}@Beanpublic RabbitTemplate rabbitTemplate(CachingConnectionFactory connectionFactory) {RabbitTemplate rabbitTemplate = new RabbitTemplate();rabbitTemplate.setConnectionFactory(connectionFactory);rabbitTemplate.setMandatory(true);rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {System.out.println("ConfirmCallback:     " + "相关数据：" + correlationData);System.out.println("ConfirmCallback:     " + "确认情况：" + ack);System.out.println("ConfirmCallback:     " + "原因：" + cause);}});rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {System.out.println("ReturnCallback:     " + "消息：" + message);System.out.println("ReturnCallback:     " + "回应码：" + replyCode);System.out.println("ReturnCallback:     " + "回应信息：" + replyText);System.out.println("ReturnCallback:     " + "交换机：" + exchange);System.out.println("ReturnCallback:     " + "路由键：" + routingKey);}});rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());return rabbitTemplate;}@Beanpublic SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(CachingConnectionFactory connectionFactory) {SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();factory.setConnectionFactory(connectionFactory);factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);return factory;}}

生产者

    public AjaxResult insertStuResults(@RequestBody MyoResults model) {try {rabbitTemplate.convertAndSend(RabbitMQConstant.EX_STU_RESULTS_MANUAL, RabbitMQConstant.Q_STU_RESULTS_MANUAL, model);return AjaxResult.success("上送成功");} catch (Exception e) {return AjaxResult.error(e.getMessage());}}

消费者

    @RabbitListener(queues = RabbitMQConstant.Q_STU_RESULTS_MANUAL)@RabbitHandlerpublic void getManualMsg(Message message, Channel channel) throws IOException {String jsonString = new String(message.getBody(), StandardCharsets.UTF_8);ObjectMapper objectMapper = new ObjectMapper();MyoResults model = objectMapper.readValue(jsonString, MyoResults.class);iStuResultsService.insertMongodb(model);channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);}