PHP高级编程之消息队列原理与实现方法详解_php技巧

来源:脚本之家  责任编辑:小易  

php基础要会,基本语法,面向对象,数据库设计对mvc框架熟悉例如 thinkphp,yii,Ci等拿thinkphp为例。假如你对thinkphp框架很熟悉了,并且能熟练应用了,然后仿照thinkphp框架 自己写一个mvc框架没有必要写的像thinkphp功能那么全,只要把核心功能写出来了就行了这个时候你会发现你的php水平会有很大的提高,差不多就达到了高级程序员了www.zgxue.com防采集请勿采集本网。

本文实例讲述了PHP高级编程之消息队列原理与实现方法。分享给大家供大家参考,具体如下:

1. 什么是消息队列

先成为初级,然后中1,中二,中三,然后就是高级了

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

2. 为什么使用消息队列

对于php,如果你以前接触过任何一种编程语言之后再学习php其实感觉php还算是相对来说比较简单的编程语言。为什么这么说呢?开源性和免费性。由于PHP的解释器的源代码是公开的,所以安全系数较高

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

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

毫无疑问是先C,一般都是C精通了之后再继续深造java、C++或者其他的,我现在也再学C,是看视频学的,我现在正在看的是夏老师的,感觉还不错,比较适合像我这样0基础的。这个比较吸引我的地方就是

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

4、C语言是基础,很重要,如果你不学好C语言,那么什么高级语言你都学不好。5、C语言与C++语言是两回事。就象大熊猫和小熊猫一样,只是名字很像!6、请先学习专业课《数据结构》、《计算机组成原理》,

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

web和android是两个方向的学习,web以后出去是做软件开发的,android以后是做app设计的,基本上能看懂,android和java是相关联的,先学java吧,希望采纳。

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

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

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

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

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

消息队列用来实现下列需求

① 存储转发

② 分布式事务

③ 发布订阅

④ 基于内容的路由

⑤ 点对点连接

5. 谁负责处理消息队列

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

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

框架化有几个好处:

① 开发者不用学习消息队列接口

② 开发者不需要关心消息推送与接收

③ 开发者通过统一的API推送消息

④ 开发者的重点是实现业务逻辑功能

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. 多线程

上面消息队列 核心代码如下

$this->queue->consume(function($envelope, $queue) { $msg = $envelope->getBody(); $result = $this->loader($msg); $queue->ack($envelope->getDeliveryTag());});

这段代码生产环境使用了半年,发现效率比较低。有些业务场入队非常快,但处理起来所花的时间就比较长,容易出现队列堆积现象。

增加多线程可能更有效利用硬件资源,提高业务处理能力。代码如下

<?phpnamespace framework;require_once( __DIR__.'/autoload.class.php' );class RabbitThread extends \Threaded { private $queue; public $classspath; protected $msg; public function __construct($queue, $logging, $msg) { $this->classspath = __DIR__.'/../queue'; $this->msg = $msg; $this->logging = $logging; $this->queue = $queue; } public function run() { $speed = microtime(true); $result = $this->loader($this->msg); $this->logging->debug('Result: '. $result.' '); $this->logging->debug('Time: '. (microtime(true) - $speed) .''); } // private public function loader($msg = null){ $protocol = json_decode($msg,true); $namespace = $protocol['Namespace']; $class = $protocol['Class']; $method = $protocol['Method']; $param = $protocol['Param']; $result = null; $classspath = $this->classspath.'/'.$this->queue.'/'.$namespace.'/'.strtolower($class) . '.class.php'; if( is_file($classspath) ){ require_once($classspath); //$class = ucfirst(substr($request_uri, strrpos($request_uri, '/')+1)); if (class_exists($class)) { if(method_exists($class, $method)){ $obj = new $class; if (!$param){ $tmp = $obj->$method(); $result = json_encode($tmp); $this->logging->info($class.'->'.$method.'()'); }else{ $tmp = call_user_func_array(array($obj, $method), $param); $result = (json_encode($tmp)); $this->logging->info($class.'->'.$method.'("'.implode('","', $param).'")'); } }else{ $this->logging->error('Object '. $class. '->' . $method. ' is not exist.'); } }else{ $msg = sprintf("Object is not exist. (%s)", $class); $this->logging->error($msg); } }else{ $msg = sprintf("Cannot loading interface! (%s)", $classspath); $this->logging->error($msg); } return $result; }}class RabbitMQ { const loop = 10; protected $queue; protected $pool; public function __construct($queueName = '', $exchangeName = '', $routeKey = '') { $this->config = new \framework\Config('rabbitmq.ini'); $this->logfile = __DIR__.'/../log/rabbitmq.%s.log'; $this->logqueue = __DIR__.'/../log/queue.%s.log'; $this->logging = new \framework\log\Logging($this->logfile, $debug=true); //.H:i:s $this->queueName = $queueName; $this->exchangeName = $exchangeName; $this->routeKey = $routeKey; $this->pool = new \Pool($this->config->get('pool')['thread']); } public function main(){ $connection = new \AMQPConnection($this->config->get('rabbitmq')); try { $connection->connect(); if (!$connection->isConnected()) { $this->logging->exception("Cannot connect to the broker!".PHP_EOL); } $this->channel = new \AMQPChannel($connection); $this->exchange = new \AMQPExchange($this->channel); $this->exchange->setName($this->exchangeName); $this->exchange->setType(AMQP_EX_TYPE_DIRECT); //direct类型 $this->exchange->setFlags(AMQP_DURABLE); //持久�? $this->exchange->declareExchange(); $this->queue = new \AMQPQueue($this->channel); $this->queue->setName($this->queueName); $this->queue->setFlags(AMQP_DURABLE); //持久�? $this->queue->declareQueue(); $this->queue->bind($this->exchangeName, $this->routeKey); $this->queue->consume(function($envelope, $queue) { $msg = $envelope->getBody(); $this->logging->debug('Protocol: '.$msg.' '); //$result = $this->loader($msg); $this->pool->submit(new RabbitThread($this->queueName,new \framework\log\Logging($this->logqueue, $debug=true), $msg)); $queue->ack($envelope->getDeliveryTag()); }); $this->channel->qos(0,1); } catch(\AMQPConnectionException $e){ $this->logging->exception($e->__toString()); } catch(\Exception $e){ $this->logging->exception($e->__toString()); $connection->disconnect(); $this->pool->shutdown(); } } private function fault($tag, $msg){ $this->logging->exception($msg); throw new \Exception($tag.': '.$msg); } public function __destruct() { }}

更多关于PHP相关内容感兴趣的读者可查看本站专题:《PHP数据结构与算法教程》、《php程序设计算法总结》、《php字符串(string)用法总结》、《PHP数组(Array)操作技巧大全》、《PHP常用遍历算法与技巧总结》及《PHP数学运算技巧总结

希望本文所述对大家PHP程序设计有所帮助。

肯定是PHP好学啊。但好学的同时,其实也等于限制了自己。而且PHP实在有点太简单了,很多现百成的方案可以直接用,已经简单到了根本不像编程。度。一个干了几年的PHP程序员,在一个其他语言的程序员眼里,几乎是“什么都不懂的小白”。JAVA像是瑞士军刀,PHP则是指甲回刀。前者几乎什么都能干,后者只能干一件事。学会使用瑞士军刀,你就要把上面每种工具都学会,自然难度也就比较大。但削水果,拧螺丝,修指甲,剪纸,都可以答用。指甲刀则简单的多,但只有修指甲这一种用途。JAVA和PHP同理,JAVA可以做网站,可以做企业管理系统,还可以做安卓开发。PHP则只能做网站内容来自www.zgxue.com请勿采集。


  • 本文相关:
  • php实现通过stomp协议连接activemq操作示例
  • php activemq的安装与使用方法图文教程
  • php使用activemq实现消息队列的方法详解
  • php使用activemq实例
  • php beanstalkd消息队列的安装与使用方法实例详解
  • php+redis实现消息队列功能示例
  • php+rabbitmq实现消息队列的完整代码
  • 使用php访问rabbitmq消息队列的方法示例
  • php实现websocket实时消息推送
  • php 使用activemq发送消息,与处理消息操作示例
  • php使用正则验证中文
  • php swoole异步mysql客户端实现方法示例
  • php动态生成指定大小随机图片的方法
  • php实现随机生成易于记忆的密码
  • php时间函数用法分析
  • php中处理mysql_fetch_assoc返回来的数组 不用foreach----echo
  • php计算十二星座的函数代码
  • php 获取目录下的图片并随机显示的代码
  • php中将网址转换为超链接的函数
  • php获取网站百度快照日期的方法
  • java和php哪个好学?
  • 如何成为PHP高级程序员??
  • 什么是MQ?
  • 电脑编程原理
  • 如何成为高级PHP开发工程师
  • 怎么自学php
  • C语言是不是一切编程的基础?包括php .net之类的?
  • 编程需要什么基础?
  • 没学web原理能看懂android4高级编程吗
  • 新手如何学习编程?
  • 网站首页网页制作脚本下载服务器操作系统网站运营平面设计媒体动画电脑基础硬件教程网络安全php基础php技巧php实例php文摘php模板首页php编程php技巧php数据结构与算法教程php程序设计算法总结php字符串(string)用法总结php数组(array)操作技巧大全php常用遍历算法与技巧总结php数学运算技巧总结php实现通过stomp协议连接activemq操作示例php activemq的安装与使用方法图文教程php使用activemq实现消息队列的方法详解php使用activemq实例php beanstalkd消息队列的安装与使用方法实例详解php+redis实现消息队列功能示例php+rabbitmq实现消息队列的完整代码使用php访问rabbitmq消息队列的方法示例php实现websocket实时消息推送php 使用activemq发送消息,与处理消息操作示例php使用正则验证中文php swoole异步mysql客户端实现方法示例php动态生成指定大小随机图片的方法php实现随机生成易于记忆的密码php时间函数用法分析php中处理mysql_fetch_assoc返回来的数组 不用foreach----echophp计算十二星座的函数代码php 获取目录下的图片并随机显示的代码php中将网址转换为超链接的函数php获取网站百度快照日期的方法php中json_decode()和json_encodphp 数组和字符串互相转换实现方php中使用curl实现get和post请求php中iconv函数使用方法php日期转时间戳,指定日期转换成php 页面跳转到另一个页面的多种php中文处理 中文字符串截取(mb_php下intval()和(int)转换使用与利用phpexcel实现excel数据的导入phpmyadmin 配置文件详解(配置)php json与xml序列化/反序列化php相对当前文件include其它文件的方法php directory 函数的详解php排序算法系列之直接选择排序详解php curl模拟登录新浪微博抓取页面内容 基修改php.ini不生效问题解决方法(上传大于php通过smtp邮件验证登陆的方法php的rsa加密解密方法以及开发接口使用php设计模式之委托模式定义与用法简单示例php判断数组元素中是否存在某个字符串的方
    免责声明 - 关于我们 - 联系我们 - 广告联系 - 友情链接 - 帮助中心 - 频道导航
    Copyright © 2017 www.zgxue.com All Rights Reserved