RabbitMQ(二) ——工作队列

RabbitMQ(二)——工作队列

(原创内容,转载请注明来源,谢谢)

一、概述

工作队列模式(work queue),是有多个消费者的情况下,可以共同消费队列内的内容,加快消息处理速度。这是RabbitMQ的基本工作模式。

二、使用方式

和上一篇中的生产和消费消息的方式一样,就是需要多在cli进程中打开一个消费者的php文件。即需要打开3个php,一个是生产者的php文件,两个消费者的php文件(或多个php文件)。

三、工作机制

3.1 轮询(Round-robin dispatching)

当开启多个生产者的时候,消费者产生消息并发送到队列的情况下,队列会将消息均衡的分发给同时打开的多个消费者。即采用轮询的方式,假设有两个消费者c1和c2,第一次有消息给c1,第二次有消息给c2,第三个再给c1,以此类推。

3.2 回馈机制(Message acknowledgment)

为了保证消息的可靠性,RabbitMQ允许用户采用消费者的ack机制,即只有消费者回馈给队列ack后,队列才会将消息从队列中剔除。这样,可以确保队列中的每个消息都是确认被消费者处理完毕的。

开启的方式,只要将方法basic_consume第四个参数设置成false,就表示开启ack机制。

开启ack,就必须要记得在消费者的代码总,加入回馈的代码,否则,消息会被队列认为没有消费,不断的堵在队列中,导致队列堵塞。

要查看队列中还没确认的内容,可以采用RabbitMQ的管理工具——rabbitmqctl。

       sudo rabbitmqctl list_queues namemessages_ready messages_unacknowledged

3.3 消息持久化(Message durability)

RabbitMQ具有完善的持久化机制,能够确保消息的安全性。可以在代码中开启持久化。消息持久化需要在队列和消息分别开启持久化。

1)队列持久化:

queue_declare的第三个参数设置为true。

2)消息持久化:

$msg = new AMQPMessage($data,
       array('delivery_mode' =>AMQPMessage::DELIVERY_MODE_PERSISTENT)
    );

这样可以确保消息可以保存在本地磁盘,因此即使rabbitmq的服务器宕机的时候,也可以保证消息的安全性。

当然,这也也存在一定的风险,因为操作系统自身的机制,开启持久化的时候,操作系统为了保证运作速度,消息会保存在操作系统层面的缓存,并且定时将消息存入硬盘。因此,还没存入磁盘的这一小段间隔(30秒左右),如果服务器宕机,有可能消息丢失。但是这个可能性很低,因此安全性已经很高。

3.4 公平分发(Fair dispatch)与预取机制(prefetch)

消息的轮询机制,每次有消息,队列则直接按照排好的顺序将消息传给对应的队列,有可能出现一部分队列上一条消息还没处理完,就出现了下一条的消息,而另外一部分队列则空闲的情况。

因此,RabbitMQ允许用户开启公平分发机制,让每个消费者只能接收一个消息,如果还没给队列回复ack,则消息来的时候,即使顺序轮到队列,也不会分发给它,而是分发给它下一个空队列。

开启方式很简单,在消费者的channel,在消费消息之前(使用basic_consume方法),给channel加一个qos机制:

       $channel->basic_qos(null, 1, null);

第二个参数就是要求消费者每次只能处理几个消息。

这样也存在一个隐患,即如果所有的消费者都还没ack,而生产者又不断的往队列发数据,则队列有可能会塞满。这个处理方案只有增加消费者,或者从代码、逻辑层面控制消息的产生速度。

——written by linhxx 2017.08.19

原文发布于微信公众号 - 决胜机器学习(phpthinker)

原文发表时间:2017-08-19

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

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏Spark学习技巧

干货:Flink+Kafka 0.11端到端精确一次处理语义实现

实时处理里消息的仅一次处理是大家关注的重点吧,前面浪尖分享过一篇对比spark streaming 和 flink的文章 <Spark Streaming VS...

753
来自专栏java技术学习之道

分布式之数据库和缓存双写一致性方案解析

1183
来自专栏性能与架构

查看操作系统负载

当我们系统有问题的时候,不要急于去调查我们代码 首先要看的是操作系统的报告,看看操作系统的CPU利用率,看看内存使用率,看看操作系统的IO,还有网络的IO,网络...

3366
来自专栏北京马哥教育

大型网站架构系列:分布式消息队列(一)

精彩早知道 消息队列概述 消息队列应用场景 消息中间件示例 JMS消息服务(见第二篇:大型网站架构系列:分布式消息队列(二)) 常用消息队列(见第二篇:大型网站...

3465
来自专栏张善友的专栏

Windows Server AppFabric:更好,更快,更便宜

微软全球副总裁Soma,负责微软公司Developer Division的工作,在博客上发布了文章Windows Server AppFabric:更好,更快,...

1768
来自专栏大数据和云计算技术

高速数据总线kafka介绍

1. Kafka的作用 在大数据系统中,常常会碰到一个问题,整个大数据是由各个子系统组成,数据需要在各个子系统中高性能,低延迟的不停流转。有没有一个系统可以同时...

3714
来自专栏张善友的专栏

开源稳定的消息队列 RabbitMQ

    RabbitMQ是一个在AMQP协议标准基础上完整的,可服用的企业消息系统。他遵循Mozilla Public License开源协议。采用 Erlan...

21110
来自专栏java一日一条

分布式系统与消息的投递

消息是一个非常有趣的概念,它是由来源发出一个离散的通信单元,被发送给一个或者一群接受者,无论是单体服务还是分布式系统中都有消息的概念,只是这两种系统中传输消息的...

603
来自专栏Java架构沉思录

缓存的正确使用方式,你都会了吗?

首先,缓存由于其适应高并发和高性能的特性,已经在项目中被广泛使用。在读取缓存方面,大家没啥疑问,都是按照下图的流程来进行业务操作。

1041
来自专栏我的博客

MYSQL主从同步

1、解决问题 数据分布不同节点、负载均衡、读写分离、容灾备份、高可用应用、故障切换等 2、同步原理 Master将操作记录到bin-log salve的一个线程...

3469

扫描关注云+社区