背景

这里后补直接上代码

最佳实践

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

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;

    @Bean
    public 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;
    }

    @Bean
    public RabbitTemplate rabbitTemplate(CachingConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public 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() {
            @Override
            public 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;
    }

    @Bean
    public 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)
    @RabbitHandler
    public 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);
    }