在上篇《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的微信公众号,我会将我的文章推送给您,并和您一起分享我日常阅读过的优质文章。

RabbitMQ入门-消息派发那些事儿的更多相关文章

  1. RabbitMQ入门-消息订阅模式

    消息派发 上篇<RabbitMQ入门-消息派发那些事儿>发布之后,收了不少反馈,其中问的最多的还是有关消息确认以及超时等场景的处理. 楼主,有遇到消费者后台进程不在,但consumer连接 ...

  2. 2.RABBITMQ 入门 - WINDOWS - 生产和消费消息 一个完整案例

    关于安装和配置,见上一篇 1.RABBITMQ 入门 - WINDOWS - 获取,安装,配置 公司有需求,要求使用winform开发这个东西(消息中间件),另外还要求开发一个日志中间件,但是也是要求 ...

  3. RabbitMQ入门教程(十七):消息队列的应用场景和常见的消息队列之间的比较

    原文:RabbitMQ入门教程(十七):消息队列的应用场景和常见的消息队列之间的比较 分享一个朋友的人工智能教程.比较通俗易懂,风趣幽默,感兴趣的朋友可以去看看. 这是网上的一篇教程写的很好,不知原作 ...

  4. RabbitMQ入门教程(十二):消息确认Ack

    原文:RabbitMQ入门教程(十二):消息确认Ack 版权声明:本文为博主原创文章,遵循CC 4.0 BY-SA版权协议,转载请附上原文出处链接和本声明. 本文链接:https://blog.csd ...

  5. RabbitMQ入门教程(十一):消息属性Properties

    原文:RabbitMQ入门教程(十一):消息属性Properties 版权声明:本文为博主原创文章,遵循CC 4.0 BY-SA版权协议,转载请附上原文出处链接和本声明. 本文链接:https://b ...

  6. RabbitMQ入门_13_消息持久化

    参考资料:https://www.rabbitmq.com/tutorials/tutorial-two-java.html 默认情况下,队列中的消息是不持久化的.如果 RabbitMQ 崩溃,队列中 ...

  7. RabbitMQ入门-高效的Work模式

    扛不住的Hello World模式 上篇<RabbitMQ入门-从HelloWorld开始>介绍了RabbitMQ中最基本的Hello World模型.正如其名,Hello World模型 ...

  8. RabbitMQ入门-Topic模式

    上篇<RabbitMQ入门-Routing直连模式>我们介绍了可以定向发送消息,并可以根据自定义规则派发消息.看起来,这个Routing模式已经算灵活的了,但是,这还不够,我们还有更加多样 ...

  9. RabbitMQ入门-从HelloWorld开始

    从读者的反馈谈RabbitMQ 昨天发完<RabbitMQ入门-初识RabbitMQ>,我陆陆续续收到一些反馈.鉴于部分读者希望结合实例来讲 期待下篇详细,最好结合案例.谢谢! 哪都好,唯 ...

随机推荐

  1. 二分法&amp;三分法

    ural History Exam    二分 #include <iostream> #include <cstdlib> using namespace std; //二分 ...

  2. Nginx图片剪裁模块探究

    http://nginx.org/en/docs/http/ngx_http_image_filter_module.html http://cwtea.blog.51cto.com/4500217/ ...

  3. selenium2 Webdriver + Java 自动化测试实战和完全教程

    selenium2 Webdriver + Java 自动化测试实战和完全教程一.快速开始 博客分类: Selenium-webdriverselenium webdriver 学习selenium ...

  4. Codeforces Round #419 (Div. 2)

    1.题目A:Karen and Morning 题意: 给出hh:mm格式的时间,问至少经过多少分钟后,该时刻为回文字符串? 思路: 简单模拟,从当前时刻开始,如果hh的回文rh等于mm则停止累计.否 ...

  5. spark2.2 DataFrame的一些算子操作

    Spark Session中的DataFrame类似于一张关系型数据表.在关系型数据库中对单表或进行的查询操作,在DataFrame中都可以通过调用其API接口来实现.可以参考,Scala提供的Dat ...

  6. 性能测试监控:Jmeter +InfluxDB +collectd +Grafana

    虚拟机ip 192.168.180.128 Influxdb Influxdb是一个开源的分布式时序.时间和指标数据库,使用go语言编写,无需外部依赖. 它有三大特性: 时序性(Time Series ...

  7. python word转pdf

    原理 使用python win32 库 调用word底层vba,将word转成pdf 安装pywin32 pip install pywin32 python代码 from win32com.clie ...

  8. vue学习之router

    路由文档:https://router.vuejs.org/zh/guide/ 使用vue做spa应用的话,一定会涉及到路由. 安装 安装router插件 npm install vue-router ...

  9. Java学习笔记30(集合框架四:List接口)

    List接口继承自Collection接口 具有重要的三大特点: 1.有序集合:存入和取出的顺序一致 2.此接口的用户可以对列表中每个元素插入位置精确的控制:可以通过索引操作 3.可以存储重复元素 L ...

  10. cygwin jdk11u

    cygwin jdk11u 安装 Cygwin64 下载地址 https://cygwin.com/setup-x86_64.exe Cygwin 国内源    中科大镜像源 http://mirro ...