[Java知识库]springBoot中使用rabbitMQ以及消息丢失问题

一、rabbitMQ中常用的交换机

图源自官网：https://www.rabbitmq.com/getstarted.html

Direct exchange 直连交换机

• 一个生产者，一个交换机，两个队列，两个消费者

根据消息发送时携带的路由routingKey(假设为mq.key.name)，将消息投递给与交换机绑定的队列(该对列与交换机绑定的路由同样为mq.key.name)

Fanout exchange 扇形交换机

• 一个生产者，一个交换机，两个队列，两个消费者

扇形交换机会将消息路由到绑定到交换机上的所有队列，可以想象成一个广播，广播消息一发出，所有听众都能够接收到消息

Topic exchange 主题交换机

• 一个生产者，一个交换机，两个队列，两个消费者

主题交换机中*跟#的区别
*代表匹配一个标识符
#可以匹配0个或者多个标识符
标识符之间由.隔开，以上图为例，现在发送一条消息message1携带路由routingKey为topic.orange.name，根据队列与交换机的路由绑定规则，messgage1将会被路由到Q1队列中，最终被C1消费者消费；发送消息message2与message3携带的路由为lazy.key和lazy.key.name,根据队列与交换机的路由绑定规则，message2和message3都会被路由到Q2队列中，最终被C2消费者消费。

Dead Letter Exchange 死信交换机

1. 先说死信队列，死信队列由死信交换机(DLX)、死信路由(DLK)、消息存活时间(TTL)组成，其中TTL为非必须部分，可以没有。
2. 当我们发送消息message4到死信队列时，我们先要有一个普通的交换机，一个普通的路由，先将死信队列绑定到普通的交换机上。当message4的存活时间到期后，message4将会被发送到普通队列queue4中(queue4通过死信路由和死信交换机绑定)，此时会被监听queue4的消费者消费。
3. 死信队列一般用在订单超时取消。

二、springBoot项目中使用rabbitMQ

创建两个项目，一个用作消息的生产者，一个用作消息的消费者
创建springBoot项目教程略过。如有疑问可自信查阅相关文档

1.两个项目中都需导入maven依赖

<!--amqp 依赖包-->
<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

两个项目中的yml文件配置基本相同

spring:
  rabbitmq:
    host: *****
    username: ****
    password: ****
    port: 5672

• host：填写rabbitmq服务的ip地址
• username：登录的用户名，如果安装时未指定，账号密码都是guest
• password：登录的密码
• port：默认是5672

yml文件的交换机、队列名称、路由名称配置

rabbit:
  applicationManualTopicExchange: mq.manual.knowledge.exchange.name
  applicationManualTopicQueue: mq.manual.knowledge.queue.name
  applicationManualRoutingKey: mq.manual.knowledge.key.name

springBoot的启动类上添加注解

@EnableRabbit

生产者项目

1.在生产者中创建RabbitMqTemplateConfig配置类

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;

/**
 * 描述 : rabbitTemplate配置、交换机、队列创建
 *
 * @AUTHOR
 * @DATE 2022/3/2 14:02
 * @DESCRIPTION:
 */
@Configuration
public class RabbitMqTemplateConfig {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    @Resource
    private CachingConnectionFactory connectionFactory;

    @Value("${rabbit.applicationManualTopicExchange}")
    private String applicationManualTopicExchange;

    @Value("${rabbit.applicationManualTopicQueue}")
    private String applicationManualTopicQueue;

    @Value("${rabbit.applicationManualRoutingKey}")
    private String applicationManualRoutingKey;

    @Bean
    public Queue applicationManualTopicQueue(){
        //true代表持久化到磁盘
        return new Queue(applicationManualTopicQueue,true);
    }
    @Bean
    public TopicExchange applicationManualTopicExchange(){
        //true代表持久化到磁盘
        return new TopicExchange(applicationManualTopicExchange,true,false);
    }
    @Bean
    public Binding manualBinding(){
        return BindingBuilder.bind(applicationManualTopicQueue()).to(applicationManualTopicExchange()).with(applicationManualRoutingKey);
    }


    @Bean
    public RabbitTemplate rabbitTemplate(){
        //connectionFactory.setPublisherConfirms(true);
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        connectionFactory.setPublisherReturns(true);
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setConfirmCallback((CorrelationData correlationData, boolean ack, String cause)->{
            logger.info("消息发送成功:correlationData({}),ack({}),cause({})",correlationData,ack,cause);
        });
        rabbitTemplate.setReturnCallback((Message message, int replyCode, String replyText, String exchange, String routingKey)->{
            logger.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}",exchange,routingKey,replyCode,replyText,message);
        });
        return rabbitTemplate;
    }

}

2.要发送的消息实体类
注意如果消息要在MQ中持久化到磁盘中必须实现Serializable接口

import java.io.Serializable;

/**
 * 描述 : 注意如果消息要在MQ中持久化到磁盘中必须实现Serializable接口
 *
 * @AUTHOR 
 * @DATE 2022/3/2 16:34
 * @DESCRIPTION:
 */
public class Message implements Serializable {

    private String messageId;

    private String content;

    public String getMessageId() {
        return messageId;
    }

    public void setMessageId(String messageId) {
        this.messageId = messageId;
    }

    public String getContent() {
        return content;
    }

    public void setContent(String content) {
        this.content = content;
    }

    @Override
    public String toString() {
        return "Message{" +
                "messageId='" + messageId + '\'' +
                ", content='" + content + '\'' +
                '}';
    }
}

3.创建生产者类BasicPublisherManager

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.***.***.***.WeChatApplicationTextMessage;
import com.***.***.***.WeChatApplicationTextMessageContent;
import com.***.***.utils.IdWorker;
import com.***.****.***.ExamPaper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.AbstractJavaTypeMapper;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;

/**
 * 描述 : 消息生产者
 *
 * @AUTHOR 
 * @DATE 2022/3/2 15:21
 * @DESCRIPTION:
 */
@Component
public class ApplicationMsgPublisherManager {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    @Resource
    private ObjectMapper objectMapper;

    @Value("${rabbit.applicationManualTopicExchange}")
    private String applicationManualTopicExchange;

    @Value("${rabbit.applicationManualRoutingKey}")
    private String applicationManualRoutingKey;

    @Resource
    private RabbitTemplate rabbitTemplate;

    @Async
    public void send(ExamPaper paper){
        WeChatApplicationTextMessage message = new WeChatApplicationTextMessage();
        WeChatApplicationTextMessageContent messageContent = new WeChatApplicationTextMessageContent();
        StringBuffer stringBuffer = new StringBuffer();
        stringBuffer.append(paper.getExamPerson());
        messageContent.setContent(String.valueOf(stringBuffer));
        //雪花算法生成的ID
        message.setId(IdWorker.getId());
        message.setToUser(paper.getPaperBy());
        message.setText(messageContent);
        sendWeChatApplicationManualMsg(message);
    }

    /**
     * 传入消息接收者，消息内容
     * @param message
     */
    public void sendWeChatApplicationManualMsg(WeChatApplicationTextMessage message){
        try {
            rabbitTemplate.setExchange(applicationManualTopicExchange);
            rabbitTemplate.setRoutingKey(applicationManualRoutingKey);
            Message msg = MessageBuilder.withBody(objectMapper.writeValueAsBytes(message))
                    .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
                    .build();
            rabbitTemplate.convertAndSend(msg);
            logger.info("基于MANUAL机制-发送消息-内容为：{} ",message);
        } catch (JsonProcessingException e) {
            logger.error("基于MANUAL机制-息转换byte异常 ",message,e.fillInStackTrace());
        } catch (Exception e){
            logger.error("基于MANUAL机制-发送消息-发生异常：{} ",message,e.fillInStackTrace());
        }
    }

    /**
     *交换机、队列、路由绑定需创建，可在config类中创建
     * 发送普通文本消息
     * @param message
     */
    public void sendStrMsg(String message){
        try {
            rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
            rabbitTemplate.setExchange("str.exchange.name");
            rabbitTemplate.setRoutingKey("str.routingKey.name");
            Message msg = MessageBuilder.withBody(message.getBytes(StandardCharsets.UTF_8))
                    .setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();
            rabbitTemplate.convertAndSend(msg);
            logger.info("文本消息发送成功，消息:{}",message);
        } catch (AmqpException e) {
            logger.error("基本消息模型-生产者-发送消息发生异常：{} ",message,e.fillInStackTrace());
        }
    }

    /**
     *交换机、队列、路由绑定需创建，可在config类中创建
     * 发送普通文本消息
     * @param message
     */
    public void sendObjectMsg(Message message){
        try {
            rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
            rabbitTemplate.setExchange("str.exchange.name");
            rabbitTemplate.setRoutingKey("str.routingKey.name");
            rabbitTemplate.convertAndSend(message,(Message msg)->{
                MessageProperties messageProperties = msg.getMessageProperties();
                messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                messageProperties.setHeader(AbstractJavaTypeMapper.DEFAULT_CONTENT_CLASSID_FIELD_NAME,Message.class);
                return msg;
            });
            logger.info("基本消息模型-生产者-发送对象类型的消息：{} ",message);
        } catch (AmqpException e) {
            logger.error("基本消息模型-生产者-发送对象类型的消息发生异常：{} ",message,e.fillInStackTrace());
        }
    }

}

消费者项目

1.创建config类

import com.***.***.***.ApplicationKnowledgeManualConsumer;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;

/**
 * 描述 :
 *
 * @AUTHOR 
 * @DATE 2022/3/2 14:02
 * @DESCRIPTION:
 */
@Configuration
public class RabbitMqTemplateConfig {

    @Resource
    private CachingConnectionFactory connectionFactory;

    @Resource
    private SimpleRabbitListenerContainerFactoryConfigurer factoryConfigurer;

    @Value("${rabbit.applicationManualTopicExchange}")
    private String applicationManualTopicExchange;

    @Value("${rabbit.applicationManualTopicQueue}")
    private String applicationManualTopicQueue;

    @Value("${rabbit.applicationManualRoutingKey}")
    private String applicationManualRoutingKey;

    /**
     * 手动确认消费者实例
     */
    @Resource
    private ApplicationKnowledgeManualConsumer applicationKnowledgeManualConsumer;


    @Bean
    public SimpleRabbitListenerContainerFactory simpleListenerContainerFactory(){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        //rabbitMQ的默认序列化方式为SerializerMessageConverter,这里我们使用Jackson2JsonMessageConverter序列化器
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        //配置多个消费者需注意消息重复消费的问题
        factory.setConcurrentConsumers(1);
        factory.setMaxConcurrentConsumers(1);
        factory.setPrefetchCount(1);
        return factory;
    }

    @Bean
    public Queue applicationManualTopicQueue(){
        //true代表持久化到磁盘
        return new Queue(applicationManualTopicQueue,true);
    }
    @Bean
    public TopicExchange applicationManualTopicExchange(){
        //true代表持久化到磁盘
        return new TopicExchange(applicationManualTopicExchange,true,false);
    }
    @Bean
    public Binding applicationManualBinding(){
        return BindingBuilder.bind(applicationManualTopicQueue()).to(applicationManualTopicExchange()).with(applicationManualRoutingKey);
    }

    /**
     * 手动确认消费消息配置
     * @param manualQueue
     * @return
     */
    @Bean
    public SimpleMessageListenerContainer simpleContainerManual(@Qualifier("applicationManualTopicQueue") Queue manualQueue){
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setMessagePropertiesConverter(new DefaultMessagePropertiesConverter());
        //TODO：并发配置，若有多个实例消费者需加锁
        container.setConcurrentConsumers(1);
        container.setMaxConcurrentConsumers(1);
        container.setPrefetchCount(1);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setQueues(manualQueue);
        //设置消费者
        container.setMessageListener(applicationKnowledgeManualConsumer);
        return container;
    }

}

2.手动确认消费者类ApplicationKnowledgeManualConsumer

import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.Channel;
import com.***.***.***.WeChatApplicationTextMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 描述 : 消费者-手动确认
 *
 * @AUTHOR 
 * @DATE 2022/3/3 11:25
 * @DESCRIPTION:
 */
@Component
public class ApplicationKnowledgeManualConsumer implements ChannelAwareMessageListener {

    private static final Logger logger= LoggerFactory.getLogger(ApplicationKnowledgeManualConsumer.class);

    @Resource
    private ObjectMapper objectMapper;

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        MessageProperties messageProperties = message.getMessageProperties();
        long deliveryTag = messageProperties.getDeliveryTag();
        try {
            byte[] messageBody = message.getBody();
            WeChatApplicationTextMessage value = objectMapper.readValue(messageBody, WeChatApplicationTextMessage.class);
            logger.info("确认消费模式-人为手动确认消费-监听器监听消费消息-内容为：{} ",value);
            //第一个参数为：消息的分发标识(唯一);第二个参数：是否允许批量确认消费(在这里设置为true)
            channel.basicAck(deliveryTag,true);
        } catch (Exception e) {
            logger.error("确认消费模式-人为手动确认消费-监听器监听消费消息-发生异常：",e.fillInStackTrace());

            //如果在处理消息的过程中发生了异常,则照样需要人为手动确认消费掉该消息 (否则该消息将一直留在队列中,从而将导致消息的重复消费)
            channel.basicReject(deliveryTag,false);
        }
    }
}

2.自动确认消费者类BasicConsumer

import com.***.***.***.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

/**
 * 描述 : 对应的交换机、队列、路由需在config类中创建
 *
 * @AUTHOR 
 * @DATE 2022/3/2 15:53
 * @DESCRIPTION:
 */
@Component
public class BasicConsumer {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    @RabbitListener(queues = "str.queue.name",containerFactory = "simpleListenerContainerFactory")
    public void consumeMsg(@Payload Message msg){
        try {
            logger.info("基本消息模型-消费者-监听消费到消息：{} ",msg);
        } catch (Exception e) {
            logger.info("基本消息模型-消费者-监听消费发生异常 ",e.fillInStackTrace());
        }
    }
}

生产者中的springBoot的测试类可参考

/**
 * 描述 :
 *
 * @AUTHOR 
 * @DATE 2022/3/2 15:41
 * @DESCRIPTION:
 */
@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest(classes = ExamApplication.class)
public class RabbitMqTest {

    @Autowired
    private BasicPublisherManager publisherManager;

    @Test
    public void test02(){
        Message message = new Message();
        message.setMessageId("1");
        message.setContent("这是一条消息");
        publisherManager.sendObjectMsg(message);
    }
}

三、消息丢失问题

生产者数据丢失

生产者向MQ发送消息的时候，可能由于网络原因殴导致消息丢失
解决方案：在生产者设置开启confirm模式(异步)后会为每一次写的消息分配一个唯一的ID，如果消息写入MQ中，MQ会给你回传一个ack消息，代表成功。相反，如果MQ没处理这个消息，会回调一个nack接口告诉你这个消息接收失败。当超过一段时间没有接收到ack消息我们可以选择重发。

MQ数据丢失

消息到MQ后，未被消费者消费，MQ挂了
解决方案：开启MQ的持久化(1.创建queue 的时候将其设置为持久化。2.在发送消息的时候将消息设置为持久化)，将消息持久化到磁盘，即使MQ挂了，恢复后会从磁盘中恢复queue，恢复这个queue里面的数据。
注意：消息到MQ后未来得及持久化到磁盘中MQ挂了也会导致数据丢失，这时候我们可以跟生产者的confirm机制结合，关闭自动ack，在数据持久化到磁盘后手动给生产者一个ack消息。

消费者数据丢失

消费者刚拿到数据还没来得及处理，消费者挂了。
解决方案：利用ack机制，在自己的代码中数据处理完之后手动ack。
注意：如果是多个消费者消费同一个消息，可能出现重复消费的问题。这时候可以结合redis，将消息id存入redis中，在redis中标识该消息是未消费还是消费中，或者消费完成。消费者消费消息前先根据消息ID去redis中查找。