专栏首页JackieZhengRabbitMQ入门-消息派发那些事儿

RabbitMQ入门-消息派发那些事儿

在上篇《RabbitMQ-高效的Work模式》中,我们了解了Work模型,该模型包括一个生产者,一个消息队列和多个消费者。 我们已经通过实例看出消息队列中的消息是如何被一个或者多个消费者消费的了,但是对于具体的实现细节和原理并没有介绍。这篇就来详细介绍下在消息派发这个过程中还有那些我们需要关注的点和细节。 这篇主要讨论细节都集中在接收端,我们还是来看下上篇中,接收端的代码实现

package com.ximalaya.openapi.rabbitmq.work;

import com.rabbitmq.client.*;

import java.io.IOException;
/**
 * Created by jackie on 17/8/4.
 */
public class Worker {

    private static final String TASK_QUEUE_NAME = "task_queue";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.3.161");
        final Connection connection = factory.newConnection();
        final Channel channel = connection.createChannel();

        channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

        channel.basicQos(1);

        final Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");

                System.out.println(" [x] Received '" + message + "'");
                try {
                    doWork(message);
                } finally {
                    System.out.println(" [x] Done");
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };
        channel.basicConsume(TASK_QUEUE_NAME, false, consumer);
    }

    private static void doWork(String task) {
        for (char ch : task.toCharArray()) {
            if (ch == '.') {
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException _ignored) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }
}

消息是怎么合理的派发给各个消费者的

在上篇介绍的实例中,我们看到运行两个消费者,这时候生产者生产的4条信息是均匀的派发给了两个消费者。

你可能会好奇,这个消息队列Queue怎么会这么“智能”,能够做到如此公平的进行消息派发。看完下面的场景你可能就不觉得RabbitMQ这样做是聪明了。

其实,默认情况下的RabbitMQ就是这么“智能”,公平、公正、公开的将4个消息依次派发给两个消费者。如果启动了四个消费者,那也将是每个消费者消费一条消息。

这是为什么呢? RabbitMQ派发消息默认采用的是轮询机制,轮询,顾名思义就是挨个的派发,就是第一个派发给C1,第二个派发给C2,第三个派发给C1,第四个派发给C4。正常情况下,这样很好,但是如果遇到某个消费者在消费某个消息时花费时间很长或者因为自身原因或者网络原因阻塞,那么按照这种轮询的策略就显得不合适了。

假设C2在执行第二个派发的消息一直卡住,这时候即使派发新的消息,C2也无法正常消费,如果一直这么盲目的派发消息给C2,只会让更多的消息无法正常消费,直至消息队列卡住崩溃。

这时候我们采用一种新的机制,姑且称为"公平机制"。该机制下,我们在同一时间内只给消费者派发一个消息(派发的数量可以人工配置),RabbitMQ只有等到该消费者确认消费了上一条消息后,才会继续派发下一条消息。

这个代码实现也很简单,就是上面接收端中的 channel.basicQos(1); 这里的数字1就是刚刚提到可以人工配置的派发消息的数量。

实例验证 要验证有basicQos和没有basicQos,我们需要做一些分析,并对代码做部分改动。

当前启动消费端,每个消费者消费的时间都是固定2秒,即使加上basicQos,因为两个队列的消费速率相同,所以最终还是会出现两个消费者各自消费两条消息的情况。

为了营造其中一个消费者卡住的情况,我们将后面启动的消费者的消费时间设定为8秒,这样第一个消费者即使消费了三条消息,这时候第二个消费者仍然卡住,便能看到效果。

下面先看没有添加basicQos的情况,第一个和第二个消费者的消费时间分别是2秒和8秒

第一个Work消费时间是2秒的,第二个是8秒

看完整个消费过程,会发现没有basiQos设置,会执行轮询策略,每个消费者都消费了两个消息

再看添加basicQos的情况,第一个和第二个消费者的消费时间同样分别是2秒和8秒

同样,第一个Work消费时间是2秒的,第二个是8秒

看完整个消费过程,会发现有basiQos设置,会执行公平机制,第一个消息给C1,第二个给C2,第三个消息来的时候,这时候发现C2还在消费,就派发给了已经消费完空闲的C1,第四个消息来的时候,发现C2仍然在消费,这时候就把消息派发给了消费完第三个消息的C1,C1总共消费3条消息用时6秒,而C2消费一条消息时8秒,所以这就是公平机制。

对比完后,我们发现这种公平机制更加合理,能够很好的做到负载均衡,避免因为不顾消费者的消费情况而盲目派发情况的出现。

如何保证派发出去的消息不丢失

现在如果出现这样的一种情况:消息从Queue中取出,但是没有消费者因为各种情况并没有完成这条消息的消费,但是这条消息已经从内存中删除了,这就意味着这个消息模型就失去了这条消息,这种意外在大多数场景下是不允许出现的。

为什么会出现这种情况呢? 因为消息出去的时候,RabbitMQ就将其从Queue中删除,也就是从内存中删除,这样做的假设前提就是默认为这条消息能够被正常消费掉,但事实情况往往并非如此。如果此时我们加上一个确认机制,类似于TCP的三次握手,问题就能够得到解决。

RabbitMQ将消息派发出去后并不立马将消息从内存中删除,等到消费端完成消费返回一个ack的标识,RabbitMQ接收到这个字段后认为消息时正常消费了在完成删除。如果没有收到确认标识ack,则认为消息违背正常消费,则会重新取回该条,采用轮询或者其他机制将其派发到下一个消费者供其消费。

实例 在接收端将代码中 channel.basicConsume(TASK_QUEUE_NAME, false, consumer); 将basicConsume函数的第二个参数改为true标识autoAck=true,即自动确认,如果设置为false,则表示需要接收端手工确认。

上篇,我们用的就是false的情况,即手动确认方式,所以在上篇的运行接口我们看到Unacknowleged标识一直从1变为0,是说明采用的是一条一条确认的机制,从第一条消息一直到第四条消息消费完成。

下面我们看看autoAck=true运行时Ready和Unacknowleged指标的变化趋势,我们只启动一个消费者

请点击此处输入图片描述 从运行过程可以发现,Unackowleged从0->4->3->2->1->0,autoAck=false是为0->1->1->1->1->0 说明autoAck=false时是一次性派发了4条信息,没有顾忌消费者是否有发送确认标识。之后消费者再依次完成消费。 ​如果您觉得阅读本文对您有帮助,请点一下“推荐”按钮,您的“推荐”将是我最大的写作动力!如果您想持续关注我的文章,请扫描二维码,关注JackieZheng的微信公众号,我会将我的文章推送给您,并和您一起分享我日常阅读过的优质文章。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • Go语言入门——数组、切片和映射

    但是对于Go,思来想去,感觉真的从“HelloWorld”说起,压根撑不住一篇的篇幅,因为Go的HelloWorld太简单了。

    JackieZheng
  • 解决Myeclipse下Debug出现Source not found以及sql server中导入数据报错

    前言:在空间里回顾了我的2014,从生活、技术、家庭等各方面对自己进行总结剖析,也是给自己一个交代。也想在博客上专门写一篇2014年度菜鸟的技术路回忆录,但是...

    JackieZheng
  • 【日拱一卒】链表——链表反转

    按照常规思维,链表反转需要知道最后一个元素,然后从最后一个元素依次往前找,直到遍历到第一个元素即完成反转。

    JackieZheng
  • XtraBackup 的流式和压缩备份

    Streaming mode, supported by Percona XtraBackup, sends backup to STDOUT in speci...

    wubx
  • js实现图片粘贴上传到服务器并展示

    最近看了一些有关于js实现图片粘贴上传的demo,实现如下: (这里只能检测到截图粘贴和图片右键复制之后粘贴) demo1: document.addEvent...

    蓓蕾心晴
  • 使用Redmine的PHP API时,如何判断需求是否为原子需求

    使用redmine的PHP接口时,怎样才能判断需求是否为原子需求呢,下面给出具体的做法:

    Enjoy233
  • 数据库无响应问题的紧急处理和分析 (r10笔记第42天)

    黄金周里处理了一起紧急的问题,在外面幸亏有同事帮忙协助,等我赶回家去,赶紧继续处理。 首先问题是在晚饭时间左右开始发生,但是过了没多久又恢复了,所以这个问题暂时...

    jeanron100
  • 智慧路灯网关下的城区路灯无线监管项目解决方案

    随着国家的繁荣昌盛以及社会文明的不断发展,作为城市重要基础设施的路灯、景观灯光已经成为体现城市形象并展示城市文明程度的重要组成部分。城市道路建设步伐的加快,路灯...

    用户7348788
  • 如何编译运行HanLP自然语言处理包

    git clone https://github.com/hankcs/HanLP.git

    IT小白龙
  • 深度学习的fine-tuning过程

    在深度学习过程中,获取数据集后,在搭建自己的网络之前需要进行的是微调,通过别人现有的网络观察自己数据的实验结果,并在此基础上,初步确定自己网络的大体结构,其中微...

    云时之间

扫码关注云+社区

领取腾讯云代金券