RabbitMQ生产端消息可靠性投递方案分析

前言

导文：
1.什么是RabbitMQ
2.Java开发技术大杂烩(三）之电商项目优化、rabbitmq、Git、OSI、VIM、Intellj IDEA、HTTP、JS、Java

之前在上面2篇文章中，讲到过RabbitMQ的安装，基本概念和用法。我们来回顾一下RabbitMQ核心基础概念。

• Server:又称之为Broker，接受客户端的连接，实现AMQP实体服务。

• Connection:连接，应用程序与Broker的网络连接。

• Channel:网络信道，几乎所有的操作都在Channel中进行，Channel是进行消息读写的通道。客户端可以建立多个Channel，每个Channel代表一个会话任务。
  如果每一次访问RabbitMQ都建立一个Connection，在消息量大的时候建立TCP Connection的开销将是巨大的，效率也较低。Channel是在connection内部建立的逻辑连接，如果应用程序支持多线程，通常每个thread创建单独的channel进行通讯，AMQP method包含了channel id帮助客户端和message broker识别channel，所以channel之间是完全隔离的。Channel作为轻量级的Connection极大减少了操作系统建立TCP connection的开销。

• Message：消息，服务器和应用程序之间传送的数据，由Message Properties和Body组成。Properties可以对消息进行修饰，比如消息的优先级，延迟等高级特性，Body就是消息体内容。

• Virtual Host:虚拟地址，用于进行逻辑隔离，最上层的消息路由。一个Virtual Host里面可以有若干个Exchange和Queue，同一个Virtual Host里面不能有相同名称的Exchange或者Queue。

• Exchange：交换机，接收消息，根据路由键转发消息到绑定的队列。

• Binding:Exchange和Queue之间的虚拟连接，binding中可以包含routing key。

• Routing key:一个路由规则，虚拟机可以用它来确定如何路由一个特定消息。

• Queue：也可以称之为Message Queue(消息队列)，保存消息并将它们转发到消费者。

通过下面2张图，我们能大概能明白AMQP协议模型和消息流转过程。在Exchange和Message Queue上面还有Virtual host。记住同一个Virtual Host里面不能有相同名称的ExChange和Message Queue。

image.png image.png

接着我们看下面的图，这是RabbitMQ消息可靠性投递的解决方案之一。

image.png

1.将消息落地到业务db和Message db。
2.采用Confirm方式发送消息至MQ Broker，返回结果的过程是异步的。Confirm消息，是指生产者投递消息后，如果Broker收到消息后，会给生产者一个ACK。生产者通过ACK，可以确认这条消息是否正常发送到Broker，这种方式是消息可靠性投递的核心。
3、4：在这里将消息分成3种状态。status=0表示消息正在投递中，status=1表示消息投递成功，status=2表示消息投递了3次还是失败。生产者接收Broker返回的Confirm确认消息结果，然后根据结果更新消息的状态。将status的状态从投递中改成投递成功即可。
5.在消息Confirm过程中，可能由于网络闪断问题或者是Broker端出现异常，导致回送消息失败或者出现异常。这时候，就需要生产者对消息进行可靠性投递，保证投递到Broker的消息可靠不丢失。还有一种极端情况值得我们考虑，那就是网络闪断。我们的消息成功投递到Broker，但是在回送ACK确认消息时，由于网络闪断，生产者没有收到。此时我们再重新投递此消息可能会造成消费端重复消费消息了。这时候需要消费端去做幂等处理(生成全局消息ID，判断此消息是否消费过)。对于没有投递成功的消息，我们可以设置一个重新投递时间。比如一个消息在5分钟内，status状态还是0，也就是这个消息还没有成功投递到Broker端。这时候我们需要一个定时任务，每隔几分钟从Message db中拉取status为0的消息。
6.将拉取的消息执行重新投递操作。
7.设置最大消息投递次数。当一个消息被投递了3次，还是不成功，那么将status置为2。最后交给人工解决处理此类问题或者将消息转存到失败表。

下面讲解一下涉及到消息可靠性的知识点和一些配置了。

application-dev.properties

#rabbtisMQ配置
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=root
spring.rabbitmq.password=root
spring.rabbitmq.virtual-host=/
#消费者数量
spring.rabbitmq.listener.simple.concurrency=10
#最大消费者数量
spring.rabbitmq.listener.simple.max-concurrency=10
#消费者每次从队列获取的消息数量
spring.rabbitmq.listener.simple.prefetch=1
#消费者自动启动
spring.rabbitmq.listener.simple.auto-startup=true
#消费失败，自动重新入队
#重试次数超过最大限制之后是否丢弃（true不丢弃时需要写相应代码将该消息加入死信队列）
#true，自动重新入队，要写相应代码将该消息加入死信队列
#false,丢弃
spring.rabbitmq.listener.simple.default-requeue-rejected=false
#是否开启消费者重试（为false时关闭消费者重试，这时消费端代码异常会一直重复收到消息）
spring.rabbitmq.listener.simple.retry.enabled=true
spring.rabbitmq.listener.simple.retry.initial-interval=1000
spring.rabbitmq.listener.simple.retry.max-attempts=3
spring.rabbitmq.listener.simple.retry.multiplier=1.0
spring.rabbitmq.listener.simple.retry.max-interval=10000

#启动发送重试策略
spring.rabbitmq.template.retry.enabled=true
#初始重试间隔为1s
spring.rabbitmq.template.retry.initial-interval=1000
#重试的最大次数
spring.rabbitmq.template.retry.max-attempts=3
#重试间隔最多10s
spring.rabbitmq.template.retry.max-interval=10000
#每次重试的因子是1.0 等差
spring.rabbitmq.template.retry.multiplier=1.0
#
#RabbitMQ的消息确认有两种。
#一种是消息发送确认。这种是用来确认生产者将消息发送给交换器，交换器传递给队列的过程中，
# 消息是否成功投递。
#发送确认分为两步，一是确认是否到达交换器，二是确认是否到达队列。
#第二种是消费接收确认。这种是确认消费者是否成功消费了队列中的消息。
# 确认消息发送成功，通过实现ConfirmCallBack接口，消息发送到交换器Exchange后触发回调
spring.rabbitmq.publisher-confirms=true
# 实现ReturnCallback接口，如果消息从交换器发送到对应队列失败时触发
# （比如根据发送消息时指定的routingKey找不到队列时会触发）
spring.rabbitmq.publisher-returns=true
# 消息消费确认，可以手动确认
spring.rabbitmq.listener.simple.acknowledge-mode=manual
#在消息没有被路由到合适队列情况下会将消息返还给消息发布者
#当mandatory标志位设置为true时，如果exchange根据自身类型和消息routingKey无法找到一个合适的queue存储消息，
# 那么broker会调用basic.return方法将消息返还给生产者;当mandatory设置为false时，
# 出现上述情况broker会直接将消息丢弃;通俗的讲，mandatory标志告诉broker代理服务器至少将消息route到一个队列中，
# 否则就将消息return给发送者;
spring.rabbitmq.template.mandatory=true

要确保RabbitMQ消息的可靠要保证以下3点：
1.publisher Confirms：要确保生产者的消息到broker的可靠性。可能会发生消息投递到broker过程中，broker挂了的情况。
2.Exchange，Queue，Message持久化：RabbitMQ是典型的内存式消息堆积。我们需要把message存储到磁盘中。如果是未持久化的消息存储在内存中，broker挂了那么消息会丢失。
3.consumer acknowledgement:消费者确认模式有3种：none(没有消息会发送应答),auto(自动应答),manual(手动应答)。为了保证消息可靠性，我们设置手动应答，这是为什么呢？采用自动应答的方式，每次消费端收到消息后，不管是否处理完成，Broker都会把这条消息置为完成，然后从Queue中删除。如果消费端消费时，抛出异常。也就是说消费端没有成功消费该消息，从而造成消息丢失。为了确保消息被消费者正确处理，我们采用手动应答(调用basicAck、basicNack、basicReject方法)，只有在消息得到正确处理下，再发送ACK。

RabbitMQ消息确认有2种：消息发送确认，消费接收确认。消息发送确认是确认生产者将消息发送到Exchange，Exchange分发消息至Queue的过程中，消息是否可靠投递。第一步是否到达Exchange，第二步确认是否到达Queue。

实现ConfirmCallBack接口,消息发送到Exchange后触发回调。

    // 消息发送到交换器Exchange后触发回调
    private final RabbitTemplate.ConfirmCallback confirmCallback =
            new RabbitTemplate.ConfirmCallback() {
                @Override
                public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                    log.info("生产端confirm...");
                    log.info("correlationData=" + correlationData);
                    String messageId = correlationData.getId();
                    if (ack) {
                        //confirm返回成功,更新消息投递状态
                        brokerMessageLogMapper.updateMessageLogStatus(messageId, Constants.ORDER_SEND_SUCCESS, new Date());
                    } else {
                        // 失败则进行具体的后续操作，重试或者补偿等手段。
                        log.info("异常处理...");
                    }
                }
            };

实现ReturnCallBack接口，消息从Exchange发送到指定的Queue失败触发回调

    // 如果消息从交换器发送到对应队列失败时触发
    private final RabbitTemplate.ReturnCallback returnCallback =
            new RabbitTemplate.ReturnCallback() {
                @Override
                public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                    log.info("message=" + message.toString());
                    log.info("replyCode=" + replyCode);
                    log.info("replyText=" + replyText);
                    log.info("exchange=" + exchange);
                    log.info("routingKey=" + routingKey);
                }
            };

消息确认机制开启，需要配置以下信息

spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.publisher-returns=true
spring.rabbitmq.listener.simple.acknowledge-mode=manual

之前说过手动应答可以调用basicAck,basicNack,basicReject方法，下面来讲讲。

手动确认消息，当multiple为false，只确认当前的消息。当multiple为true，批量确认所有比当前deliveryTag小的消息。deliveryTag是用来标识Channel中投递的消息。RabbitMQ保证在每个Channel中，消息的deliveryTag是从1递增。

image.png

当消费端处理消息异常时，我们可以选择处理失败消息的方式。如果requeue为true，失败消息会重新进入Queue，试想一下，如果消费者在消费时发生异常，那么就不会对这一次消息进行ACK，进而发生回滚消息的操作，使消息始终放在Queue的头部，然后不断的被处理和回滚，导致队列陷入死循环，为了解决这种问题，我们可以引入重试机制(当重试次数超过最大值，丢弃该消息)或者是死信队列+重试队列。
requeue为false，丢弃该消息。

image.png

和basicNack用法一样。

image.png

为了配合Return机制，我们要配置spring.rabbitmq.template.mandatory=true。它的作用是在消息没有被路由到合适的队列情况下，Broker会将消息返回给生产者。当mandatory为true时，如果Exchange根据类型和消息Routing Key无法路由到一个合适的Queue存储消息，那么Broker会调用Basic.Return回调给handleReturn()，再回调给ReturnCallback，将消息返回给生产者。当mandatory为false时，丢弃该消息。

    @Override
    public void handleReturn(int replyCode,
            String replyText,
            String exchange,
            String routingKey,
            BasicProperties properties,
            byte[] body)
        throws IOException {

        ReturnCallback returnCallback = this.returnCallback;
        if (returnCallback == null) {
            Object messageTagHeader = properties.getHeaders().remove(RETURN_CORRELATION_KEY);
            if (messageTagHeader != null) {
                String messageTag = messageTagHeader.toString();
                final PendingReply pendingReply = this.replyHolder.get(messageTag);
                if (pendingReply != null) {
                    returnCallback = new ReturnCallback() {

                        @Override
                        public void returnedMessage(Message message, int replyCode, String replyText, String exchange,
                                String routingKey) {
                            pendingReply.returned(new AmqpMessageReturnedException("Message returned",
                                    message, replyCode, replyText, exchange, routingKey));
                        }
                    };
                }
                else if (logger.isWarnEnabled()) {
                    logger.warn("Returned request message but caller has timed out");
                }
            }
            else if (logger.isWarnEnabled()) {
                logger.warn("Returned message but no callback available");
            }
        }
        if (returnCallback != null) {
            properties.getHeaders().remove(PublisherCallbackChannel.RETURN_CORRELATION_KEY);
            MessageProperties messageProperties = this.messagePropertiesConverter.toMessageProperties(
                    properties, null, this.encoding);
            Message returnedMessage = new Message(body, messageProperties);
            returnCallback.returnedMessage(returnedMessage,
                    replyCode, replyText, exchange, routingKey);
        }
    }

当消息路由不到合适的Queue，会在回调给ReturnCallck这些信息。

image.png

如果消费端忘记了ACK，这些消息会一直处于Unacked 状态。由于RabbitMQ消息消费没有超时机制，也就是程序不重启，消息会一直处于Unacked状态。当消费端程序关闭时，这些处于Unack状态的消息会重新恢复成Ready状态。这时候会出现一种情况：当消费端程序开启时，由于Broker端积压了大量的消息，又可能会让消费端崩溃。所以我们要对消费端进行限流处理。RabbitMQ提供了一种qos(Quality of Service,服务质量保证)功能，即在非自动ACK前提下，如果一定数量的消息未被ACK前，不进行新消息的消息。

image.png

spring.rabbitmq.listener.simple.prefetch=1

image.png

下面贴消息可靠性解决方案代码了。

配置任务调度中心

@Configuration
@EnableScheduling
public class TaskSchedulerConfig implements SchedulingConfigurer {

    protected ThreadPoolExecutor threadPoolExecutor;

    @Override
    public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
        taskRegistrar.setScheduler(taskExecutor());
    }

    @Bean(destroyMethod = "shutdown")
    public ThreadPoolExecutor taskExecutor() {
        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
                .setNameFormat("task-executor-pool-%d").build();
        this.threadPoolExecutor = new ScheduledThreadPoolExecutor(10, namedThreadFactory,
                new ThreadPoolExecutor.AbortPolicy());
        return threadPoolExecutor;
    }

}

执行重新投递status为0的消息。这里也可以使用corn表达式设置触发任务调度的时间。关于fixedRate和fixedDelay概念总有人搞混。fixedRate任务两次执行时间间隔是任务的开始点，而fixedDelay的间隔是前次任务的结束和下一次任务开始的间隔。

@Component
@Slf4j
public class RetryMessageTask {

    @Autowired
    private RabbitmqOrderSender rabbitmqOrderSender;

    @Autowired
    private BrokerMessageLogMapper brokerMessageLogMapper;

    @Scheduled(initialDelay = 5000, fixedDelay = 30000)
    public void trySendMessage() {
        log.info("定时投递status为0的消息...");
        List<BrokerMessageLog> brokerMessageLogList = brokerMessageLogMapper.listStatusAndTimeoutMessage();
        brokerMessageLogList.forEach(brokerMessageLog -> {
            if (brokerMessageLog.getTryCount() >= 3) {
                log.info("投递3次还是失败...");
                brokerMessageLogMapper.updateMessageLogStatus(brokerMessageLog.getMessageId(),
                        Constants.ORDER_SEND_FAIL,
                        new Date());
            } else {
                log.info("投递失败...");
                brokerMessageLogMapper.updateReSendMessage(brokerMessageLog.getMessageId(),
                        new Date());
                Order order = JSON.parseObject(brokerMessageLog.getMessage(), Order.class);
                try {
                    rabbitmqOrderSender.sendOrder(order);
                } catch (Exception e) {
                    log.error("重新投递消息发送异常...:" + e.getMessage());
                }
            }
        });
    }
}

消息生产端

@Component
@Slf4j
public class RabbitmqOrderSender {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private BrokerMessageLogMapper brokerMessageLogMapper;

    // 消息发送到交换器Exchange后触发回调
    private final RabbitTemplate.ConfirmCallback confirmCallback =
            new RabbitTemplate.ConfirmCallback() {
                @Override
                public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                    log.info("生产端confirm...");
                    log.info("correlationData=" + correlationData);
                    String messageId = correlationData.getId();
                    if (ack) {
                        //confirm返回成功,更新消息投递状态
                        brokerMessageLogMapper.updateMessageLogStatus(messageId, Constants.ORDER_SEND_SUCCESS, new Date());
                    } else {
                        // 失败则进行具体的后续操作，重试或者补偿等手段。
                        log.info("异常处理...");
                    }
                }
            };

    // 如果消息从交换器发送到对应队列失败时触发
    private final RabbitTemplate.ReturnCallback returnCallback =
            new RabbitTemplate.ReturnCallback() {
                @Override
                public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                    log.info("message=" + message.toString());
                    log.info("replyCode=" + replyCode);
                    log.info("replyText=" + replyText);
                    log.info("exchange=" + exchange);
                    log.info("routingKey=" + routingKey);
                }
            };

    public void sendOrder(Order order) {
        log.info("生产端发送消息...");
        rabbitTemplate.setConfirmCallback(this.confirmCallback);
        rabbitTemplate.setReturnCallback(this.returnCallback);
        CorrelationData correlationData = new CorrelationData(order.getMessageId());
        rabbitTemplate.convertAndSend(MQConfig.ORDER_DIRECT_EXCAHNGE,
                MQConfig.ORDER_QUEUE,  order, correlationData);
    }
}

消息消费端

@Component
@Slf4j
public class RabbitmqOrderReceiver {

    @RabbitListener(queues = MQConfig.ORDER_QUEUE)
    public void receive(@Payload Order order, Channel channel,
                        @Headers Map<String, Object> headers,
                        Message message) throws IOException, InterruptedException {
        log.info("消费端接收消息...");
        log.info("message=" + message.toString());
        log.info("order=" + order);
        Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
        log.info("deliveryTag=" + deliveryTag);
        // 手工ack
        channel.basicAck(deliveryTag, false);
    }
}

当我们发送消息时，故意将Exchange设置成一个不存在的值。消息路由不到合适的Exchange，Confirm机制回送的ACK会返回false，走异常处理。这个消息的状态不会更新成1。然后定时任务会拉取status为0的消息，进行重新投递，投递了3次消息还未成功，将status置为2。

image.png

接下来，我们测试一波。

    @Test
    public void test() {
        Order order = new Order();
        order.setId("36");
        order.setName("cmazxiaoma测试订单-36");
        order.setMessageId(UUIDUtil.uuid());

        rabbitmqOrderService.createOrder(order);
    }

消息投递失败。

image.png

定时任务重新投递消息失败。

image.png

将失败的消息重新投递3次还是失败。

image.png

更新Message db信息，将重新投递3次还是失败的消息状态置为2。

image.png

接着我们把消费端手动ACK的代码注释掉，再让生产端发送消息。看看会出现什么情况。

image.png

我们会发现Queue堆积了该消息。

image.png

我们关掉RabbitMQ Server，看看此消息是否会持久化。

[root@VM_0_11_centos log]# ps -ef|grep rabbitmq
root     13283 10291  0 13:42 pts/1    00:00:00 grep --color=auto rabbitmq
root     23051     1  1 Nov06 ?        00:09:29 /usr/lib64/erlang/erts-5.10.4/bin/beam -W w -A 64 -P 1048576 -t 5000000 -stbt db -zdbbl 128000 -K true -- -root /usr/lib64/erlang -progname erl -- -home /root -- -pa /usr/local/rabbitmq/ebin -noshell -noinput -s rabbit boot -sname rabbit@VM_0_11_centos -boot start_sasl -kernel inet_default_connect_options [{nodelay,true}] -sasl errlog_type error -sasl sasl_error_logger false -rabbit error_logger {file,"/usr/local/rabbitmq/var/log/rabbitmq/rabbit@VM_0_11_centos.log"} -rabbit sasl_error_logger {file,"/usr/local/rabbitmq/var/log/rabbitmq/rabbit@VM_0_11_centos-sasl.log"} -rabbit enabled_plugins_file "/usr/local/rabbitmq/etc/rabbitmq/enabled_plugins" -rabbit plugins_dir "/usr/local/rabbitmq/plugins" -rabbit plugins_expand_dir "/usr/local/rabbitmq/var/lib/rabbitmq/mnesia/rabbit@VM_0_11_centos-plugins-expand" -os_mon start_cpu_sup false -os_mon start_disksup false -os_mon start_memsup false -mnesia dir "/usr/local/rabbitmq/var/lib/rabbitmq/mnesia/rabbit@VM_0_11_centos" -kernel inet_dist_listen_min 25672 -kernel inet_dist_listen_max 25672 -noshell -noinput
[root@VM_0_11_centos log]# kill -9 23051

[root@VM_0_11_centos sbin]# rabbitmq-server -detached
Warning: PID file not written; -detached was passed.
[root@VM_0_11_centos sbin]# ps -ef|grep rabbitmq
root     13500     1 31 13:44 ?        00:00:02 /usr/lib64/erlang/erts-5.10.4/bin/beam -W w -A 64 -P 1048576 -t 5000000 -stbt db -zdbbl 128000 -K true -- -root /usr/lib64/erlang -progname erl -- -home /root -- -pa /usr/local/rabbitmq/ebin -noshell -noinput -s rabbit boot -sname rabbit@VM_0_11_centos -boot start_sasl -kernel inet_default_connect_options [{nodelay,true}] -sasl errlog_type error -sasl sasl_error_logger false -rabbit error_logger {file,"/usr/local/rabbitmq/var/log/rabbitmq/rabbit@VM_0_11_centos.log"} -rabbit sasl_error_logger {file,"/usr/local/rabbitmq/var/log/rabbitmq/rabbit@VM_0_11_centos-sasl.log"} -rabbit enabled_plugins_file "/usr/local/rabbitmq/etc/rabbitmq/enabled_plugins" -rabbit plugins_dir "/usr/local/rabbitmq/plugins" -rabbit plugins_expand_dir "/usr/local/rabbitmq/var/lib/rabbitmq/mnesia/rabbit@VM_0_11_centos-plugins-expand" -os_mon start_cpu_sup false -os_mon start_disksup false -os_mon start_memsup false -mnesia dir "/usr/local/rabbitmq/var/lib/rabbitmq/mnesia/rabbit@VM_0_11_centos" -kernel inet_dist_listen_min 25672 -kernel inet_dist_listen_max 25672 -noshell -noinput
root     13597 10291  0 13:44 pts/1    00:00:00 grep --color=auto rabbitmq

执行rabbitmqctl list_queues name messages_ready messages_unacknowledged命令，查询Queue情况，发现Message持久化了。

image.png
image.png

断开消费者程序，我们可以看到消息从Unacked状态转换成Ready了。

image.png

尾言

不管是神还是恶魔都不会对不抗争的人伸出援手