文档章节

RabbitMQ与PHP(一)

j
 jims
发布于 2017/03/28 13:23
字数 1916
阅读 12
收藏 1
点赞 0
评论 0

RabbitMQ与PHP(一)

项目中使用RabbitMQ作为队列处理用户消息通知,消息由前端PHP代码产生,处理消息使用Python,这就导致代码一致性问题,调整消息定义时需要PHP和Python都进行修改。这两天抽时间研究了下,如何将消息的产生与处理(消费)全部用PHP来做。查资料时发现,关于PHP处理消息队列的资料很少,有必要把一些初学者容易混淆的地方再讲一下。

拟分成两部分: 一,RabbitMQ的原理与操作示例;二,具体服务安装及如何用PHP作为守护模式处理消息。

RabbitMQ是流行的开源消息队列系统,用erlang语言开发,完整的实现了AMPQ(高级消息队列协议)。网站在: http://www.rabbitmq.com/ 上面有教程和实例代码(Python和Java的)。

AMPQ协议为了能够满足各种消息队列需求,在概念上比较复杂。首先,rabbitMQ启动默认是没有任何配置的,需要客户端连接上去,设置交换机等才能工作。不把这些基础概念弄清楚,后面程序设计就容易产生问题。

1,vhosts : 虚拟主机

一个RabbitMQ的实体上可以有多个vhosts,用户与权限设置就是依附于vhosts。对一般PHP应用,不需要用户权限设定,直接使用默认就存在的"/"就可以了,用户可以使用默认就存在的"guest"。一个简单的配置示例:

$conn_args = array( 
    'host' => '127.0.0.1',  
    'port' => '5672',  
    'login' => 'guest',  
    'password' => 'guest', 
    'vhost'=>'/'
);

2,connection 与 channel : 连接与信道

connection是指物理的连接,一个client与一个server之间有一个连接;一个连接上可以建立多个channel,可以理解为逻辑上的连接。一般应用的情况下,有一个channel就够用了,不需要创建更多的channel。示例代码:

//创建连接和channel 
$conn = new AMQPConnection($conn_args);   
if (!$conn->connect()) {   
    die("Cannot connect to the broker!\n");   
}   
$channel = new AMQPChannel($conn);  
##3,exchange 与  routingkey : 交换机 与 路由键##

为了将不同类型的消息进行区分,设置了交换机与路由两个概念。比如,将A类型的消息发送到名为‘C1’的交换机,将类型为B的发送到'C2'的交换机。当客户端连接C1处理队列消息时,取到的就只是A类型消息。进一步的,如果A类型消息也非常多,需要进一步细化区分,比如某个客户端只处理A类型消息中针对K用户的消息,routingkey就是来做这个用途的。

$e_name = 'e_linvo'; //交换机名 
$k_route = array(0=> 'key_1', 1=> 'key_2'); //路由key 
//创建交换机    
$ex = new AMQPExchange($channel);   
$ex->setName($e_name); 
$ex->setType(AMQP_EX_TYPE_DIRECT); //direct类型  
$ex->setFlags(AMQP_DURABLE); //持久化 
echo "Exchange Status:".$ex->declare()."\n";   
for($i=0; $i<5; ++$i){ 
    echo "Send Message:".$ex->publish($message . date('H:i:s'), $k_route[i%2])."\n";  
}

由以上代码可以看到,发送消息时,只要有“交换机”就够了。至于交换机后面有没有对应的处理队列,发送方是不用管的。routingkey可以是空的字符串。在示例中,我使用了两个key交替发送消息,是为了下面更便于理解routingkey的作用。

对于交换机,有两个重要的概念:

A,类型。有三种类型: Fanout类型最简单,这种模型忽略routingkey;Direct类型是使用最多的,使用确定的routingkey。这种模型下,接收消息时绑定'key_1'则只接收key_1的消息;最后一种是Topic,这种模式与Direct类似,但是支持通配符进行匹配,比如: 'key_*',就会接受key_1和key_2。Topic貌似美好,但是有可能导致不严谨,所以还是推荐使用Direct。

B,持久化。指定了持久化的交换机,在重新启动时才能重建,否则需要客户端重新声明生成才行。

需要特别明确的概念:交换机的持久化,并不等于消息的持久化。只有在持久化队列中的消息,才能持久化;如果没有队列,消息是没有地方存储的;消息本身在投递时也有一个持久化标志的,PHP中默认投递到持久化交换机就是持久的消息,不用特别指定。

4,queue: 队列

讲了这么多,才讲到队列呀。事实上,队列仅是针对接收方(consumer)的,由接收方根据需求创建的。只有队列创建了,交换机才会将新接受到的消息送到队列中,交换机是不会在队列创建之前的消息放进来的。换句话说,在建立队列之前,发出的所有消息都被丢弃了。下面这个图比RabbitMQ官方的图更清楚——Queue是属于ReceiveMessage的一部分。

接下来看一下创建队列及接收消息的示例:

$e_name = 'e_linvo'; //交换机名 
$q_name = 'q_linvo'; //队列名 
$k_route = ''; //路由key 

//创建连接和channel 
$conn = new AMQPConnection($conn_args);   
if (!$conn->connect()) {   
    die("Cannot connect to the broker!\n");   
}   
$channel = new AMQPChannel($conn);   

//创建交换机    
$ex = new AMQPExchange($channel);   
$ex->setName($e_name); 
$ex->setType(AMQP_EX_TYPE_DIRECT); //direct类型  
$ex->setFlags(AMQP_DURABLE); //持久化 
echo "Exchange Status:".$ex->declare()."\n";   

//创建队列    
$q = new AMQPQueue($channel); 
$q->setName($q_name);   
$q->setFlags(AMQP_DURABLE); //持久化  

//绑定交换机与队列,并指定路由键 
echo 'Queue Bind: '.$q->bind($e_name, $k_route)."\n"; 

//阻塞模式接收消息 
echo "Message:\n";   
$q->consume('processMessage', AMQP_AUTOACK); //自动ACK应答  

$conn->disconnect();   

/**
 * 消费回调函数
 * 处理消息
 */
function processMessage($envelope, $queue) { 
    var_dump($envelope->getRoutingKey);
    $msg = $envelope->getBody(); 
    echo $msg."\n"; //处理消息 
}

从上述示例中可以看到,交换机既可以由消息发送端创建,也可以由消息消费者创建。

创建一个队列(line:20)后,需要将队列绑定到交换机上(line:25)队列才能工作,routingkey也是在这里指定的。有的资料上写成bindingkey,其实一回事儿,弄两个名词反倒容易混淆。

消息的处理,是有两种方式:

A,一次性。用 $q->get([...]),不管取到取不到消息都会立即返回,一般情况下使用轮询处理消息队列就要用这种方式;

B,阻塞。用 $q->consum( callback, [...] ) 程序会进入持续侦听状态,每收到一个消息就会调用callback指定的函数一次,直到某个callback函数返回FALSE才结束。

关于callback,这里多说几句: PHP的call_back是支持使用数组的,比如: $c = new MyClass(); $c->counter = 100; $q->consume( array($c,'myfunc') ) 这样就可以调用自己写的处理类。MyClass中myfunc的参数定义,与上例中processMessage一样就行。

在上述示例中,使用的$routingkey = '', 意味着接收全部的消息。我们可以将其改为 $routingkey = 'key_1',可以看到结果中仅有设置routingkey为key_1的内容了。

注意: routingkey = 'key_1' 与 routingkey = 'key_2' 是两个不同的队列。假设: client1 与 client2 都连接到 key_1 的队列上,一个消息被client1处理之后,就不会被client2处理。而 routingkey = '' 是另类,client_all绑定到 '' 上,将消息全都处理后,client1和client2上也就没有消息了。

在程序设计上,需要规划好exchange的名称,以及如何使用key区分开不同类型的标记,在消息产生的地方插入发送消息代码。后端处理,可以针对每一个key启动一个或多个client,以提高消息处理的实时性。如何使用PHP进行多线程的消息处理,将在下一节中讲述。

更多消息模型,可以参考: http://www.rabbitmq.com/tutorials/tutorial-two-python.html

本文转载自:

共有 人打赏支持
j
粉丝 4
博文 109
码字总数 29707
作品 0
合肥
CentOS7.X安装RabbitMQ-3.6.10

CentOS7.0安装RabbitMQ 安装前的准备 源码安装erlang erlang下载 erlang-20.0 加入环境变量 测试启动erlang 安装RabbitMQ 下载地址 Binary .tar.xz .zip cd /rootwget http://www.rabbitmq.c...

qq2233466866 ⋅ 06/11 ⋅ 0

RabbitMQ 内存控制 硬盘控制

一、内存控制: vmmemoryhigh_watermark 该值为内存阈值,默认为0.4。意思为物理内存的40%。40%的内存并不是内存的最大的限制,它是一个发布的节制,当达到40%时Erlang会做GC。最坏的情况是使...

andrewniu ⋅ 05/10 ⋅ 0

RabbitMQ 3.7.5-beta.3 发布,带来多处 bug 修复

RabbitMQ 3.7.5-beta.3 发布,此版本是维护版本的预览版,主要带来了多处 bug 修复,涉及模块包括: Core Server CLI Tools Management Plugin LDAP Plugin Shovel Plugin Peer Discovery A...

雨田桑 ⋅ 04/30 ⋅ 0

RabbitMQ 3.7.5-rc.1 发布,bug 修复版本

RabbitMQ 3.7.5-rc.1 发布,此版本是维护版本的候选版本,主要是对一些 bug 进行了修复。 更新涉及模块包括: Core Server CLI Tools Management Plugin Federation Plugin LDAP Plugin Shov...

雨田桑 ⋅ 05/04 ⋅ 0

rabbitmq-server 安装

一,安装rabbitmq-server 1.安装erlang wget https://packages.erlang-solutions.com/erlang-solutions-1.0-1.noarch.rpm rpm -Uvh erlang-solutions-1.0-1.noarch.rpm rpm --import https:/......

丿小贰丶 ⋅ 05/08 ⋅ 0

消息中间件—RabbitMQ(初探篇)

文章摘要:本篇文章为RabbitMQ的入门文章,不像其他一些程序代码和应用实战性的文章会带着大家从一个“Hello World”的简单例子出发,在该篇幅中主要给大家讲下RabbitMQ消息队列的起源、为何...

癫狂侠 ⋅ 05/23 ⋅ 0

消息中间件—RabbitMQ(集群监控篇1)

摘要:任何没有监控的系统上线,一旦在生产环境发生故障,那么排查和修复问题的及时性将无法得到保证 一、为何要对消息中间件进行监控? 上线的业务系统需要监控,然而诸如消息队列、数据库、...

癫狂侠 ⋅ 05/28 ⋅ 0

RabbitMQ 3.7.6-rc.2 发布,AMQP 消息服务器

RabbitMQ 3.7.6-rc.2 已发布,这是常规维护版本的候选版,主要包括错误修复。 本次更新包含如下内容: Core Server Bug Fixes Max priority cap for queues is now enforced and set to 255...

淡漠悠然 ⋅ 06/05 ⋅ 0

Docker下RabbitMQ三部曲之一:极速体验(单机和集群)

从本章开始,我们一起在Docker环境实战RabbitMQ环境部署和对应的Java开发,当前是《Docker下RabbitMQ三部曲》系列的第一篇,整个三部曲由以下三篇文章组成: 1. 第一篇,即本章,我们用最快的...

boling_cavalry ⋅ 05/12 ⋅ 0

RabbitMQ消息中间件介绍

一、基础介绍 随着分布式应用的发展消息队列中间件成为C/S架构中解耦的一个重要环节,传统的消息传输模型中,C端发出消息,S端必须在线,否则将无法继续进行,而在拥有消息中间件的模型下消息...

qiao645 ⋅ 前天 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

PHP语言系统ZBLOG或许无法重现月光博客的闪耀历史[图]

最近在写博客,希望通过自己努力打造一个优秀的教育类主题博客,名动江湖,但是问题来了,现在写博客还有前途吗?面对强大的自媒体站点围剿,还有信心和可能型吗? 至于程序部分,我选择了P...

原创小博客 ⋅ 15分钟前 ⋅ 0

IntelliJ IDEA 2018.1新特性

工欲善其事必先利其器,如果有一款IDE可以让你更高效地专注于开发以及源码阅读,为什么不试一试? 本文转载自:netty技术内幕 3月27日,jetbrains正式发布期待已久的IntelliJ IDEA 2018.1,再...

Romane ⋅ 40分钟前 ⋅ 0

浅谈设计模式之工厂模式

工厂模式(Factory Pattern)是 Java 中最常用的设计模式之一。这种类型的设计模式属于创建型模式,它提供了一种创建对象的最佳方式。 在工厂模式中,我们在创建对象时不会对客户端暴露创建逻...

佛系程序猿灬 ⋅ 今天 ⋅ 0

Dockerfile基础命令总结

FROM 指定使用的基础base image FROM scratch # 制作base image ,不使用任何基础imageFROM centos # 使用base imageFROM ubuntu:14.04 尽量使用官方的base image,为了安全 LABEL 描述作...

ExtreU ⋅ 昨天 ⋅ 0

存储,对比私有云和公有云的不同

导读 说起公共存储,很难不与后网络公司时代的选择性外包联系起来,但尽管如此,它还是具备着简单和固有的可用性。公共存储的名字听起来也缺乏专有性,很像是把东西直接堆放在那里而不会得到...

问题终结者 ⋅ 昨天 ⋅ 0

C++难点解析之const修饰符

C++难点解析之const修饰符 c++ 相比于其他编程语言,可能是最为难掌握,概念最为复杂的。结合自己平时的C++使用经验,这里将会列举出一些常见的难点并给出相应的解释。 const修饰符 const在c...

jackie8tao ⋅ 昨天 ⋅ 0

聊聊spring cloud netflix的HystrixCommands

序 本文主要研究一下spring cloud netflix的HystrixCommands。 maven <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-clo......

go4it ⋅ 昨天 ⋅ 0

Confluence 6 从其他备份中恢复数据

一般来说,Confluence 数据库可以从 Administration Console 或者 Confluence Setup Wizard 中进行恢复。 如果你在恢复压缩的 XML 备份的时候遇到了问题,你还是可以对整个站点进行恢复的,如...

honeymose ⋅ 昨天 ⋅ 0

myeclipse10 快速搭建spring boot开发环境(入门)

1.创建一个maven的web项目 注意上面标红的部分记得选上 2.创建的maven目录结构,有缺失的目录可以自己建立目录补充 补充后 这时候一个maven的web项目创建完成 3.配置pom.xml配置文件 <proje...

小海bug ⋅ 昨天 ⋅ 0

nginx.conf

=========================================================================== nginx.conf =========================================================================== user nobody; #......

A__17 ⋅ 昨天 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部