文档章节

PHP高级编程之消息队列

netkiller-
 netkiller-
发布于 2015/10/19 15:21
字数 1780
阅读 938
收藏 29
点赞 6
评论 1

PHP高级编程之消息队列

http://netkiller.github.io/journal/php.mq.html

Mr. Neo Chen (陈景峰), netkiller, BG7NYT


中国广东省深圳市龙华新区民治街道溪山美地
518131
+86 13113668890
+86 755 29812080
<netkiller@msn.com>

版权声明

转载请与作者联系,转载时请务必标明文章原始出处和作者信息及本声明。

文档出处:
http://netkiller.github.io
http://netkiller.sourceforge.net

微信扫描二维码进入 Netkiller 微信订阅号

QQ群:128659835 请注明“读者”

2015-10-19

摘要
我的系列文档
Netkiller Architect 手札 Netkiller Developer 手札 Netkiller PHP 手札 Netkiller Python 手札 Netkiller Testing 手札
Netkiller Cryptography 手札 Netkiller Linux 手札 Netkiller Debian 手札 Netkiller CentOS 手札 Netkiller FreeBSD 手札
Netkiller Shell 手札 Netkiller Security 手札 Netkiller Web 手札 Netkiller Monitoring 手札 Netkiller Storage 手札
Netkiller Mail 手札 Netkiller Docbook 手札 Netkiller Project 手札 Netkiller Database 手札 Netkiller PostgreSQL 手札
Netkiller MySQL 手札 Netkiller NoSQL 手札 Netkiller LDAP 手札 Netkiller Network 手札 Netkiller Cisco IOS 手札
Netkiller H3C 手札 Netkiller Multimedia 手札 Netkiller Perl 手札 Netkiller Amateur Radio 手札 Netkiller DevOps 手札

您可以使用iBook阅读当前文档


1. 什么是消息队列

消息队列(英语:Message queue)是一种进程间通信或同一进程的不同线程间的通信方式

2. 为什么使用消息队列

消息队列技术是分布式应用间交换信息的一种技术。消息队列可驻留在内存或磁盘上,队列存储消息直到它们被应用程序读出。通过消息队列,应用程序可独立地执行,它们不需要知道彼此的位置、或在继续执行前不需要等待接收程序接收此消息。

3. 什么场合使用消息队列

你首先需要弄清楚,消息队列与远程过程调用的区别,在很多读者咨询我的时候,我发现他们需要的是RPC(远程过程调用),而不是消息队列。

消息队列有同步或异步实现方式,通常我们采用异步方式使用消息队列,远程过程调用多采用同步方式。

MQ与RPC有什么不同? MQ通常传递无规则协议,这个协议由用户定义并且实现存储转发;而RPC通常是专用协议,调用过程返回结果。

4. 什么时候使用消息队列

同步需求,远程过程调用(PRC)更适合你。

异步需求,消息队列更适合你。

目前很多消息队列软件同时支持RPC功能,很多RPC系统也能异步调用。

消息队列用来实现下列需求
  1. 存储转发

  2. 分布式事务

  3. 发布订阅

  4. 基于内容的路由

  5. 点对点连接

5. 谁负责处理消息队列

通常的做法,如果小的项目团队可以有一个人实现,包括消息的推送,接收处理。如果大型团队,通常是定义好消息协议,然后各自开发各自的部分, 例如一个团队负责写推送协议部分,另一个团队负责写接收与处理部分。

那么为什么我们不讲消息队列框架化呢?

框架化有几个好处:
  1. 开发者不用学习消息队列接口
  2. 开发者不需要关心消息推送与接收
  3. 开发者通过统一的API推送消息
  4. 开发者的重点是实现业务逻辑功能

6. 怎么实现消息队列框架

下面是作者开发的一个SOA框架,该框架提供了三种接口,分别是SOAP,RESTful,AMQP(RabbitMQ),理解了该框架思想,你很容易进一步扩展,例如增加XML-RPC, ZeroMQ等等支持。

https://github.com/netkiller/SOA

本文只讲消息队列框架部分。

6.1. 守护进程

消息队列框架是本地应用程序(命令行程序),我们为了让他在后台运行,需要实现守护进程。

https://github.com/netkiller/SOA/blob/master/bin/rabbitmq.php

每个实例处理一组队列,实例化需要提供三个参数,$queueName = '队列名', $exchangeName = '交换名', $routeKey = '路由'

$daemon = new \framework\RabbitDaemon($queueName = 'email', $exchangeName = 'email', $routeKey = 'email');

守护进程需要使用root用户运行,运行后会切换到普通用户,同事创建进程ID文件,一边进程停止的时候使用。

守护进程核心代码https://github.com/netkiller/SOA/blob/master/system/rabbitdaemon.class.php

6.2. 消息队列协议

消息协议是一个数组,将数组序列化或者转为JSON推送到消息队列服务器,这里使用json格式的协议。

$msg = array(
	'Namespace'=>'namespace',
	"Class"=>"Email",
	"Method"=>"smtp",
	"Param" => array(
		$mail, $subject, $message, null
	)
);

序列化后的协议

{"Namespace":"single","Class":"Email","Method":"smtp","Param":["netkiller@msn.com","Hello"," TestHelloWorld",null]}

使用json格式是考虑到通用性,这样推送端可以使用任何语言。如果不考虑兼容,建议使用二进制序列化,例如msgpack效率更好。

6.3. 消息队列处理

消息队列处理核心代码

https://github.com/netkiller/SOA/blob/master/system/rabbitmq.class.php

所以消息的处理在下面一段代码中进行

$this->queue->consume(function($envelope, $queue) {

	$speed = microtime(true);
	
	$msg = $envelope->getBody();
	$result = $this->loader($msg);
	$queue->ack($envelope->getDeliveryTag()); //手动发送ACK应答

	//$this->logging->info(''.$msg.' '.$result)
	$this->logging->debug('Protocol: '.$msg.' ');
	$this->logging->debug('Result: '. $result.' ');
	$this->logging->debug('Time: '. (microtime(true) - $speed) .'');
});

public function loader($msg = null) 负责拆解协议,然后载入对应的类文件,传递参数,运行方法,反馈结果。

Time 可以输出程序运行所花费的时间,对于后期优化十分有用。

提示

loader() 可以进一步优化,使用多线程每次调用loader将任务提交到线程池中,这样便可以多线程处理消息队列。

6.4. 测试

测试代码 https://github.com/netkiller/SOA/blob/master/test/queue/email.php

<?php
$queueName = 'example';
$exchangeName = 'email';
$routeKey = 'email';
$mail = $argv[1];
$subject = $argv[2];
$message = empty($argv[3]) ? 'Hello World!' : ' '.$argv[3];

 
$connection = new AMQPConnection(array(
	'host' => '192.168.4.1', 
	'port' => '5672', 
	'vhost' => '/', 
	'login' => 'guest', 
	'password' => 'guest'
	));
$connection->connect() or die("Cannot connect to the broker!\n");
 
$channel = new AMQPChannel($connection);
$exchange = new AMQPExchange($channel);
$exchange->setName($exchangeName);
$queue = new AMQPQueue($channel);
$queue->setName($queueName);
$queue->setFlags(AMQP_DURABLE);
$queue->declareQueue();

$msg = array(
	'Namespace'=>'namespace',
	"Class"=>"Email",
	"Method"=>"smtp",
	"Param" => array(
		$mail, $subject, $message, null
	)
);

$exchange->publish(json_encode($msg), $routeKey);
printf("[x] Sent %s \r\n", json_encode($msg));


$connection->disconnect();

这里只给出了少量测试与演示程序,如有疑问请到渎者群,或者公众号询问。

7. 总结

该消息队列框架还比较简陋,但在生产环境已经运行很长一段时间,效果还是不错的。同时降低了消息队列的开发难度,开发者更多的时间是考虑业务逻辑的实现,而不用操心消息队列本身的使用。

© 著作权归作者所有

共有 人打赏支持
netkiller-

netkiller-

粉丝 667
博文 240
码字总数 322337
作品 10
深圳
部门经理
加载中

评论(1)

swon
swon
赞一个!79
服务器攻防·禁止穷举邮箱密码

服务器攻防·禁止穷举邮箱密码 原创 2016-11-02 景峯 Netkiller 本文节选自《Netkiller Linux 手札》 下面是一个脚本你可以放在crontab中运行,每个一定时间运行一次。 #!/bin/bash Homepage...

neo-chen ⋅ 2016/11/02 ⋅ 0

Shell 高级编程

Shell 高级编程 原创 2016-11-04 景峯 Netkiller Shell 高级编程 http://netkiller.github.io/journal/shell.html Mr. Neo Chen (陈景峯), netkiller, BG7NYT 中国广东省深圳市龙华新区民治街......

neo-chen ⋅ 2016/11/04 ⋅ 0

PHP结合JAVA的开发技术--LAJP

LAJP名称来源于著名的LAMP(Linux,Apache,Mysql,Php),LAMP是轻量级的开发 Web程序的环境,在Internet上有广泛的应用,但对于企业开发,如金融、电信领域,LAMP显得能力不足,这些领域通常是J...

刁忆飞 ⋅ 2010/03/30 ⋅ 0

分布式异步编程框架--Koper

Koper Koper是一个基于消息和事件驱动的分布式异步编程框架, Koper的设计目标是为大型场景提供高性能高吞吐的简单编程方案。 Koper为分布式环境提供了简化的消息监听和数据事件监听模型,它...

raymondkk ⋅ 2016/11/18 ⋅ 1

Php调用java说明

Php调用java说明 (lajp实现) LAJP名称含义 LAJP名称来源于著名的LAMP(Linux,Apache,Mysql,Php),LAMP是轻量级的开发Web程序的环境,在Internet上有广泛的应用,但对于企业开发,如金融、电...

史帝文 ⋅ 2016/11/05 ⋅ 0

PHP实现的AMQP--php-amqplib

php-amqplib是Advanced Message Queuing Protocol (AMQP)的一个PHP开源实现。 高级消息队列协议(AMQP)是一个异步消息传递所使用的应用层协议规范。作为线路层协议,而不是API(例如JMS),...

匿名 ⋅ 2010/11/21 ⋅ 0

PHP的异步并行网络扩展Swoole已发布1.7.5版本

PHP的异步并行网络扩展Swoole今天发布了最新的1.7.5版本。 项目主页:http://www.swoole.com/ 文档页面:http://wiki.swoole.com/ 源代码:https://github.com/swoole/swoole-src 1.7.4版本:...

matyhtf ⋅ 2014/09/10 ⋅ 31

RabbitMQ官方中文入门教程(PHP版) 第一部分:Hello World

RabbitMQ是一个消息代理。它的核心原理非常简单:接收和发送消息。你可以把它想像成一个邮局:你把信件放入邮箱,邮递员就会把信件投递到你的收件人处。在这个比喻中,RabbitMQ是一个邮箱、邮...

tree2013 ⋅ 2016/11/11 ⋅ 0

PHP高级编程之守护进程,实现优雅重启

PHP高级编程之守护进程 http://netkiller.github.io/journal/php.daemon.html Mr. Neo Chen (陈景峰), netkiller, BG7NYT 中国广东省深圳市龙华新区民治街道溪山美地 518131 +86 13113668890...

neo-chen ⋅ 2015/10/20 ⋅ 17

(六)RabbitMQ消息队列-消息任务分发与消息ACK确认机制(PHP版)

在前面一章介绍了在PHP中如何使用RabbitMQ,至此入门的的部分就完成了,我们内心中一定还有很多疑问:如果多个消费者消费同一个队列怎么办?如果这几个消费者分任务的权重不同怎么办?怎么把...

Super_RD ⋅ 2017/04/26 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

005. 深入JVM学习—Java堆内存参数调整

1. JVM整体内存调整图解(调优关键) 实际上每一块子内存区域都会存在一部分可变伸缩区域,其基本流程:如果内存空间不足,则在可变的范围之内扩大内存空间,当一段时间之后,内存空间不紧张...

影狼 ⋅ 10分钟前 ⋅ 0

内存障碍: 软件黑客的硬件视图

此文为笔者近日有幸看到的一则关于计算机底层内存障碍的学术论文,并翻译(机译)而来[自认为翻译的还行],若读者想要英文原版的论文话,给我留言,我发给你。 内存障碍: 软件黑客的硬件视图...

Romane ⋅ 43分钟前 ⋅ 0

SpringCloud 微服务 (七) 服务通信 Feign

壹 继续第(六)篇RestTemplate篇 做到现在,本机上已经有注册中心: eureka, 服务:client、order、product 继续在order中实现通信向product服务,使用Feign方式 下面记录学习和遇到的问题 贰 or...

___大侠 ⋅ 今天 ⋅ 0

gitee、github上issue标签方案

目录 [TOC] issue生命周期 st=>start: 开始e=>end: 结束op0=>operation: 新建issueop1=>operation: 评审issueop2=>operation: 任务负责人执行任务cond1=>condition: 是否通过?op3=>o......

lovewinner ⋅ 今天 ⋅ 0

浅谈mysql的索引设计原则以及常见索引的区别

索引定义:是一个单独的,存储在磁盘上的数据库结构,其包含着对数据表里所有记录的引用指针. 数据库索引的设计原则: 为了使索引的使用效率更高,在创建索引时,必须考虑在哪些字段上创建索...

屌丝男神 ⋅ 今天 ⋅ 0

String,StringBuilder,StringBuffer三者的区别

这三个类之间的区别主要是在两个方面,即运行速度和线程安全这两方面。 首先说运行速度,或者说是, 1.执行速度 在这方面运行速度快慢为:StringBuilder(线程不安全,可变) > StringBuffer...

时刻在奔跑 ⋅ 今天 ⋅ 0

java以太坊开发 - web3j使用钱包进行转账

首先载入钱包,然后利用账户凭证操作受控交易Transfer进行转账: Web3j web3 = Web3j.build(new HttpService()); // defaults to http://localhost:8545/Credentials credentials = Wallet......

以太坊教程 ⋅ 今天 ⋅ 0

Oracle全文检索配置与实践

Oracle全文检索配置与实践

微小宝 ⋅ 今天 ⋅ 0

mysql的分区和分表

1,什么是mysql分表,分区 什么是分表,从表面意思上看呢,就是把一张表分成N多个小表,具体请看mysql分表的3种方法 什么是分区,分区呢就是把一张表的数据分成N多个区块,这些区块可以在同一...

梦梦阁 ⋅ 今天 ⋅ 0

exception.ZuulException: Forwarding error

错误日志 com.netflix.zuul.exception.ZuulException: Forwarding error Caused by: com.netflix.hystrix.exception.HystrixRuntimeException: xxx timed-out and no fallback available. Ca......

jack_peng ⋅ 今天 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部