PHP消息队列详解

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了PHP消息队列详解，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含8559字，纯文字阅读大概需要13分钟。

内容图文

本文主要和大家分享PHP消息队列详解，希望能帮助到大家，首先我们先了解一下什么是消息队列。

1. 什么是消息队列

消息队列（英语：Message queue）是一种进程间通信或同一进程的不同线程间的通信方式

2. 为什么使用消息队列

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

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

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

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

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());
});

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

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

<?php
namespace 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消息队列详解所遇到的程序开发问题。如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容，请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端

扫描二维码推送至手机访问。

本文链接：https://qyyshop.com/info/138554.html

来源：【匿名】

【上一篇】PHP实现消息队列【下一篇】PHP 消息队列 Kafka 使用

更多 ►

【PHP消息队列详解】教程文章相关的互联网学习教程文章

配置文件<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:aop="http://www.springframework.org/schema/aop"xmlns:tx="http://www.springframework.org/schema/tx" xmlns:util="http://www.springframework.org/schema/util"xmlns:context="http://www.springframework.org/schema/context" xmlns:mongo="http:/...

ActiveMQ消息队列使用和配置【代码】【图】

安装ActiveMQ使用brew安装https://blog.csdn.net/u010046908/article/details/54728375直接下载安装https://blog.csdn.net/ytangdigl/article/details/77740100启动activemq start帐号密码都是admin默认端口8161 搭建项目框架这边拿一个简易的spring+springMVC为框架添加pom依赖 <dependency><groupId>org.activiti</groupId><artifactId>activiti-engine</artifactId><version>${activiti.version}</...

Linux消息队列【代码】

1 #include <stdio.h> 2 #include <sys/types.h> 3 #include <sys/ipc.h> 4 #include <sys/msg.h> 5 #include <errno.h> 6 #include <string.h>7 #include <stdlib.h>8 9int msg_key = 1234; 1011struct MsgNode 12{ 13long nMsgType; 14char sBuf[256]; 15}; 1617int main(int argc,char** argv) 18{ 19int nMsgId = 0; 20if((argc == 2) && (strcmp(argv[1],"c") == 0)) 21 { 22 nMsgId = msgget(msg_key,I...

消息队列汇总（转）

add by zhj: 文中不仅列出了最常用的MQ，而且还列出了一些文章，分享使用的经验，值得一看原文：http://queues.io/AboutThere are many queueing systems out there. Each one of them is different and was created for solving certain problems. This page tries to collect the libraries that are widely popular and have a successful record of running on (big) production systems.The goal is to create a quality list...

消息队列简介

一、概述计算机科学中，消息队列和邮箱是用于进程间或者线程与同一进行间通讯的软件工程组件。他们都是消息传传输控制队列。消息队列是发布/订阅模型的变种，是较大的面向消息的中间件的一部分。多数消息系统支持发布/订阅和消息队列模型的API，如JMS（Java Message Service）。消息队列提供异步的通讯协议，这就意味着消息发送者和消息接收者不需要在同一时间与消息队列交互。消息入队直到接收者来读取。消息队列都有单条消息大小...

利用消息队列实现ECHO_SRV【代码】

server端代码：server 从队列的mtype=1接收数据在发到mtype=pid（client的进程id）#include <stdio.h> #include <stdlib.h> #include <string.h> #include <sys/types.h> #include <sys/ipc.h> #include <sys/msg.h> #include <errno.h> #define MSGMAX 8192 #define ERR_EXIT(m) do { perror(m); exit(EXIT_FAILURE); }while(0)struct msgbuf{long mtype;char mtext[MSGMAX]; //消息缓冲区};void echo_s...

关于消息队列中间件选型和线程同步问题

消息队列中间件选型推荐使用ZeroMQZeroMQ：号称最快最轻量级的消息队列系统，尤其是大吞吐量的需求场景。专门为高吞吐量/低延迟的场景开发，金融项目使用比较多。MSMQ：微软集成的消息队列，寄宿在Window服务里，性能稳定,优于ActiveMQ，但是无法跨平台,。ActiveMQ：类似于ZemoMQ,跨平台，它易于实现高级场景，而且只需付出低消耗，它可以部署于代理模式和P2P模式，ActiveMQ被誉为Java世界的中坚力量，但是跑不过MSMQ，需要安装Jav...

【消息队列高手课】- 进阶篇【代码】【图】

Kafka的官方文档：http://kafka.apache.org/documentation/#quickstartKafka的核心论文：https://www.kancloud.cn/kancloud/log-real-time-datas-unifying/58708Kafka原理介绍文档：https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Improvement+Proposals、http://kafka.apache.org/documentation/#design、http://kafka.apache.org/documentation/#implementation--代码的异步化处理CompletableFuture简单介绍：runAsyn...

SpringBoot消息队列之-rabbitMQ【代码】【图】

一、概述1.在大多应用中，我们系统之间需要进行异步通信，即异步消息。2.异步消息中两个重要概念：消息代理（message broker）和目的地（destination）当消息发送者发送消息以后，将由消息代理接管，消息代理保证消息传递到指定目的地。3.异步消息主要有两种形式的目的地队列（queue）：点对点消息通信（point-to-point）主题（topic）：发布（publish）/订阅（subscribe）消息通信4.点对点式：–消息发送者发送消息，消息代理将其...

.NET消息队列的使用【代码】【图】

在项目中使用消息队列，可以对消息提供方和消息消费方进行解耦，提高系统的灵活性和可拓展性。其中，企业级的消息队列中间件有IBM WebSphere MQ、Tibco EMS、HornetQ等等。微软的Windows系统也自带了消息队列的功能，可以在“启动或关闭Windows功能”中开启：开启后在服务中查看消息队列服务是否运行：接下来是简单的消息队列操作：using System; using System.Messaging;namespace MQ {class MQHelper{///<summary>/// 创建一个消...

消息队列的使用【图】

一、消息队列概述消息队列中间件是分布式系统中重要的组件，主要解决应用解耦，异步消息，流量削锋等问题，实现高性能，高可用，可伸缩和最终一致性架构。目前使用较多的消息队列有ActiveMQ，RabbitMQ，ZeroMQ，Kafka，MetaMQ，RocketMQ二、消息队列应用场景以下介绍消息队列在实际应用中常用的使用场景。异步处理，应用解耦，流量削锋和消息通讯四个场景。2.1异步处理场景说明：用户注册后，需要发注册邮件和注册短信。传统的做法...

Kafka介绍与消息队列【图】

消息队列的好处：消息队列（Message Queue）消息: 网络中的两台计算机或者两个通讯设备之间传递的数据。例如说：文本、音乐、视频等内容。队列：一种特殊的线性表（数据元素首尾相接），特殊之处在于只允许在首部删除元素和在尾部追加元素。入队、出队。消息队列：顾名思义，消息+队列，保存消息的队列。消息的传输过程中的容器；主要提供生产、消费接口供外部调用做数据的存储和获取。消息队列分类MQ分类：点对点（P2P）、发布订阅...

MSMQ(消息队列)【图】

前段时间研究WCF接触到了MSMQ，所以认真的学习了一下，下面是我的笔记。我理解的MSMQ MSMQ可以被看成一个数据储存装置，就如同数据库，只不过数据存储的是一条一条的记录，而MSMQ存储的是一个一个的消息(messsge)。Message可以被理解为一种数据容器，我们在稍后会讲到。MSMQ一个重要的应用场景就是离线信息交互，例如，我们在给朋友发送邮件，而此时朋友并未登入邮箱，这个时候我们的邮件就可以发到邮件服务器的MSMQ队列中，当...

使用Zabbix监控RabbitMQ消息队列【图】

注：本文档根据GitHub开源项目，实践成功后所编写开源项目地址：https://github.com/jasonmcintosh/rabbitmq-zabbix为方便大家下载，已经将此项目整个打包上传至附件，可从附件中直接下载监控部署督步骤：下载附件中文件复制项目中scripts目录中所有内容至 zabbix_agentd 端服务器的 /usr/local/sbin 目录下（任何包含 zabbix_agentd 端服务器均可）注：放置文件的服务器需要能与rabbitmq服务器通讯，假设我存放的服务器为agent1上...

进程间通信——消息队列【图】

概念消息队列提供了一种从一个进程向另一个进程发送一个数据块的方法，每个数据块都被认为是有一个类型，接收者进程接受的数据块可以有不同的类型值。我们可以通过发送消息来避免命名管道的同步和阻塞问题。消息队列与管道不同的是，消息队列是基于消息的，而管道是基于字节流的，且消息队列的读取不一定是先入先出。消息队列与命名管道有一样的不足，就是每个消息的最大长度是有上限的（MSGMAX），每个消息队列的总字节数是有上限...

消息队列 - 相关标签

消息队列

消息队列 - 最热教程

rabbitmq连接数量一直增多 RabbitMQ如何避免消息重复投递或重复消...【MQ中间件】RabbitMQ -- RabbitMQ死信...rabbitMQ无法启动解决 Springboot接入RabbitMQ详细教程 linux（debian）系统中安装RabbitMQ Caused by:com.rabbitmq.client.Shutdo...Docker搭建RabbitMQ集群(镜像模式)阿里云linux服务器上部署系统记录（jdk...linux 消息队列

首页 / 消息队列 / PHP消息队列详解

PHP消息队列详解

内容导读

内容图文

内容总结

内容备注

内容手机端

【PHP消息队列详解】教程文章相关的互联网学习教程文章

使用redis的发布订阅模式实现消息队列【代码】

ActiveMQ消息队列使用和配置【代码】【图】

Linux消息队列【代码】

消息队列汇总（转）

消息队列简介

利用消息队列实现ECHO_SRV【代码】

关于消息队列中间件选型和线程同步问题

【消息队列高手课】- 进阶篇【代码】【图】

SpringBoot消息队列之-rabbitMQ【代码】【图】

.NET消息队列的使用【代码】【图】

消息队列的使用【图】

Kafka介绍与消息队列【图】

MSMQ(消息队列)【图】

使用Zabbix监控RabbitMQ消息队列【图】

进程间通信——消息队列【图】

消息队列 - 相关标签

PHP - 相关标签

消息队列 - 最新教程

消息队列 - 最热教程