文档章节

PHP高级编程之消息队列

netkiller-
 netkiller-
发布于 2015/10/19 15:21
字数 1780
阅读 1.2K
收藏 31

精选30+云产品,助力企业轻松上云!>>>

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-

粉丝 715
博文 274
码字总数 379578
作品 10
深圳
部门经理
私信 提问
加载中
此博客有 1 条评论,请先登录后再查看。
服务器攻防·禁止穷举邮箱密码

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

neo-chen
2016/11/02
53
0
软件-MQ-RabbitMQ:RabbitMQ

ylbtech-软件-MQ-RabbitMQ:RabbitMQ RabbitMQ是实现了高级消息队列协议(AMQP)的开源消息代理软件(亦称面向消息的中间件)。RabbitMQ服务器是用Erlang语言编写的,而集群和故障转移是构建...

osc_499wyr2d
2019/05/28
27
0
php从业者不同阶段必须掌握的职业技能

PHP初级开发工程师 PHP初级开发工程师需要掌握的技能 1. 走进PHP的世界 PHP介绍、PHP版本发展历史、PHP薪资水平和职业发展、WEB开发中基础概念介绍、PHP开发环境搭建、HTML+CSS等入门内容。 ...

小志yang
2018/06/08
13
0
RabbitMQ的基本概念与原理(一)

1、什么是MQ? MQ全称为Message Queue, 顾名思义,即消息队列,它是一种应用程序对应用程序的通信方法。MQ是消费-生产者模型的一个典型的代表,生产者不断的往消息队列中不断写入消息,消费者...

osc_veu13fnv
2019/08/03
4
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
420
0

没有更多内容

加载失败,请刷新页面

加载更多

限时免费在线学习考取百度智能云ABC(初级)在线认证证书

本文作者:y****n 百度云智学院致力于为百度ABC战略(人工智能、大数据、云计算)提供人才生态体系建设,包括基于百度ABC、IoT的课程体系,整合百度优势技术能力的深度学习技术、Apollo无人车...

百度开发者中心
昨天
5
0
Maven中dependencyManagement和依赖关系之间的区别

问题: What is the difference between dependencyManagement and dependencies ? dependencyManagement和dependencies什么区别? I have seen the docs at Apache Maven web site. 我已经在......

富含淀粉
今天
7
0
Confluence 如何查看页面 ID

如果你希望查看页面的 ID 你有 2 个方法。 例如,你希望查看 https://www.cwiki.us/display/CONFLUENCEWIKI/Get+started 页面的 Page ID 的话。 如果你的标题栏没有特殊字符,那么将会使用英...

honeymoose
今天
11
0
Linux系统 Centos7 环境基于Docker部署Rocketmq服务

消息队列 基本概述 MQ,Message Queue,基于TCP协议构建的简单协议,区别于具体的通信协议。 基于通信协议定义和抽象的更高层次的通信模型,一般都是生产者和消费者模型,又或者说服务端和客户端...

mazhilin
今天
12
0
我可以在.gitconfig中为自己指定多个用户吗? - Can I specify multiple users for myself in .gitconfig?

问题: In my ~/.gitconfig , I list my personal email address under [user] , since that's what I want to use for Github repos. 在~/.gitconfig ,我在[user]我的个人电子邮件地址,因......

法国红酒甜
今天
19
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部