PHP中RabbitMQ之phpAmqplib实现(五
环境:CoentOS,PHP 7
简单介绍一下php-amqplib
php-amqplib是Advanced Message Queuing Protocol (AMQP)的一个PHP开源实现。高级消息队列协议(AMQP)是一个异步消息传递所使用的应用层协议规范。作为线路层协议,而不是API(例如JMS),AMQP 客户端能够无视消息的来源任意发送和接受信息
1、RabbitMQ的安装
需要下载的两个包
erlang-21.0.7-1.el7.centos.x86_64.rpm
rabbitmq-server-3.7.7-1.el7.noarch.rpm
这两个包我已经放在了百度云盘的分享上
链接:https://pan.baidu.com/s/1rMv_yFpLnH-D1S5wrOZrbA#list/path=%2FRabbitMQ
提取码:ipyu
然后参照 weixin_41368339的博客linux rabbitmq3.7.7安装与使用一文中的步骤安装步,基本上没有什么问题
2、composer的安装(已安装的请忽略此步)
为什么要装这个?我们可以通过composer来下载安装php-amqplib
如何安装composer,可以百度一下composer的全局安装或者直接去composer中文网
3、php-amqplib的下载及安装
新建一个composer.json的文件,内容如下所示
-
{
-
"require": {
-
"php-amqplib/php-amqplib": ">=2.6.1"
-
}
-
}
然后执行
composer install
会生成一个composer.lock文件及vendor文件夹,vendor文件夹里有php-amqplib库,且有一个autoload.php文件可以使用自动加载
4、Demo示例
本Demo示例只创建了一个直连交换机,共有四个文件Consumer.php (消费者),Publisher.php (生产者) ,Parenter.php (自己封装的RabbitMQ的方法) ,以及test.php (测试数据),目录如图所示
Parenter.php 代码如下图所示
-
-
require_once __DIR__ . '/vendor/autoload.php';
-
-
use PhpAmqpLib\Connection\AMQPStreamConnection;
-
use PhpAmqpLib\Message\AMQPMessage;
-
abstract class Parenter
-
{
-
//MQ的默认连接配置
-
public $config = array(
-
'host' => '127.0.0.1', //ip
-
'port' => '5672', //端口号
-
'user' => 'guest', //用户
-
'password' => 'guest', //密码
-
'vhost' => '/' //虚拟host
-
);
-
-
public $connection; //链接
-
public $channel; //信道
-
-
public $exchangeName = ''; //交换机名
-
public $queueName = ''; //队列名
-
public $routeKey = ''; //路由键
-
public $exchangeType = 'direct'; //交换机类型
-
-
public $autoAck = true; //是否自动ack应答
-
-
public function __construct($exchangeName, $queueName, $routeKey, $exchangeType = 'direct', $config=array())
-
{
-
$this->exchangeName = empty($exchangeName) ? '' : $exchangeName;
-
$this->queueName = empty($queueName) ? '' : $queueName;
-
$this->routeKey = empty($routeKey) ? '' : $routeKey;
-
$this->exchangeType = empty($exchangeType) ? '' : 'direct';
-
if(!empty($config))
-
{
-
$this->setConfig($config);
-
}
-
$this->createConnect();
-
}
-
-
//创建连接与信道
-
private function createConnect()
-
{
-
$host = $this->config['host'];
-
$port = $this->config['port'];
-
$user = $this->config['user'];
-
$password = $this->config['password'];
-
$vhost = $this->config['vhost'];
-
if(empty($host) || empty($port) || empty($user) || empty($password))
-
{
-
throw new Exception('RabbitMQ的连接配置不正确');
-
}
-
//创建链接
-
$this->connection = new AMQPStreamConnection($host, $port, $user, $password, $vhost);
-
//创建信道
-
$this->channel = $this->connection->channel();
-
$this->createExchange();
-
}
-
-
//创建交换机
-
private function createExchange()
-
{
-
//创建交换机$channel->exchange_declare($exhcange_name,$type,$passive,$durable,$auto_delete);
-
//passive: 消极处理, 判断是否存在队列,存在则返回,不存在直接抛出 PhpAmqpLib\Exception\AMQPProtocolChannelException 异常
-
//durable:true、false true:服务器重启会保留下来Exchange。警告:仅设置此选项,不代表消息持久化。即不保证重启后消息还在
-
//autoDelete:true、false.true:当已经没有消费者时,服务器是否可以删除该Exchange
-
$this->channel->exchange_declare($this->exchangeName, $this->exchangeType, false, true, false);
-
//passive: 消极处理, 判断是否存在队列,存在则返回,不存在直接抛出 PhpAmqpLib\Exception\AMQPProtocolChannelException 异常
-
//durable:true、false true:在服务器重启时,能够存活
-
//exclusive :是否为当前连接的专用队列,在连接断开后,会自动删除该队列
-
//autodelete:当没有任何消费者使用时,自动删除该队列
-
//arguments: 自定义规则
-
$this->channel->queue_declare($this->queueName, false, true, false, false);
-
}
-
-
//发送消息
-
public function sendMessage($data)
-
{
-
//创建消息$msg = new AMQPMessage($data,$properties)
-
//#$data string类型 要发送的消息
-
//#roperties array类型 设置的属性,比如设置该消息持久化[‘delivery_mode’=>2]
-
$msg = new AMQPMessage($data, array('delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT));
-
$this->channel->basic_publish($msg,$this->exchangeName, $this->routeKey);
-
}
-
-
//处理消息
-
public function dealMq($flag)
-
{
-
$this->autoAck = $flag;
-
$this->channel->queue_bind($this->queueName,$this->exchangeName, $this->routeKey);
-
//prefetchSize:0
-
//prefetchCount:会告诉RabbitMQ不要同时给一个消费者推送多于N个消息,即一旦有N个消息还没有ack,则该consumer将block掉,直到有消息ack
-
//global:true\false 是否将上面设置应用于channel,简单点说,就是上面限制是channel级别的还是consumer级别
-
//$this->channel->basic_qos(0, 1, false);
-
//1:queue 要取得消息的队列名
-
//2:consumer_tag 消费者标签
-
//3:no_local false这个功能属于AMQP的标准,但是rabbitMQ并没有做实现.参考
-
//4:no_ack false收到消息后,是否不需要回复确认即被认为被消费
-
//5:exclusive false排他消费者,即这个队列只能由一个消费者消费.适用于任务不允许进行并发处理的情况下.比如系统对接
-
//6:nowait false不返回执行结果,但是如果排他开启的话,则必须需要等待结果的,如果两个一起开就会报错
-
//7:callback null回调函数
-
//8:ticket null
-
//9:arguments null
-
$this->channel->basic_consume($this->queueName, '', false, $this->autoAck, false, false, function($msg){$this->get($msg);});
-
//监听消息
-
while(count($this->channel->callbacks)){
-
$this->channel->wait();
-
}
-
}
-
-
public function get($msg)
-
{
-
$param = $msg->body;
-
$this->doProcess($param);
-
if(!$this->autoAck)
-
{
-
//手动ack应答
-
$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
-
}
-
}
-
-
abstract public function doProcess($param);
-
-
public function closeConnetct()
-
{
-
$this->channel->close();
-
$this->connection->close();
-
}
-
-
//重新设置MQ的链接配置
-
public function setConfig($config)
-
{
-
if (!is_array($config))
-
{
-
throw new Exception('config不是一个数组');
-
}
-
foreach($config as $key => $value)
-
{
-
$this->config[$key] = $value;
-
}
-
}
-
}
Consumer.php (消费者)
-
-
include_once('Parenter.php');
-
class Consumer extends Parenter
-
{
-
public function __construct()
-
{
-
parent::__construct('exchange', 'queue', 'routeKey');
-
}
-
public function doProcess($msg)
-
{
-
echo $msg."\n";
-
}
-
}
-
$consumer = new Consumer();
-
//$consumer->dealMq(false);
-
$consumer->dealMq(true);
Publisher.php (生产者)
-
-
include_once('Parenter.php');
-
class Publisher extends Parenter
-
{
-
public function __construct()
-
{
-
parent::__construct('exchange', '', 'routeKey');
-
}
-
public function doProcess($msg)
-
{
-
-
}
-
-
}
test.php(测试数据)
-
-
include_once('Publisher.php');
-
$publisher = new Publisher();
-
$publisher->sendMessage('Hello,World!');
-
$publisher->closeConnetct();
5、添加交换机与队列
打开http://ip(你的RabbitMQ安装的主机):15672/,会进入到RabbitMQ的可视化管理后台登录页面,登录你的账号密码(如果你是按照第一步提到的博客里的教程来装的,那你的账号密码就是guest),然后新加交换机和队列,
以下是新加交换机的操作,注意vhost与以及交换机的名称要与代码里的消费者与生产者传入的参数值保持一致,如果你不想使用"/"这个默认的vhost,也可以新建一个vhost(什么?你问我如何新建,那么请百度一下),但是要记住在代码里创建消费者与生产者时把你新加的这个vhost传进去,覆盖RabbitMqParernt.php里的vhost
以下是新加队列,这里的vhost要与上一步的vhost保持一致,保证交换机与队列在同一个vhost下,不然交换机会找不到队列的,队列名与消费者代码里传入进去的队列名保持一致
6、运行代码
先打开一个窗口启动消费者
运行测试脚本
如果打印出来字符串就成功了
注意:消费者与生产者传入的交换机名称,路由键必须相同
交换机类型请务必选择直连,各种交换机的路由键形式不大相同,有兴趣的同学可以去试试其它类型的交换机实现哦
当修改了vhost或者交换机名称,队列名称等时,需要修改对应代码
至于注释里的ack应答,我会在之后的博客里详细介绍,包括RabbitMQ的持久化,这里使用默认的ack应答即可
代码里很多注释都是我后来学习php-amqplib库中类的方法时加的,表示的是参数的意义,大家也可以去研究一下,这里提供个网址: Rabbitmq各方法的作用详解
关于管理后台及RabbitMQ的命令,我这里就不多介绍了,有兴趣的同学去网上搜索一下就能搜到好多
下一篇:RabbitMQ的持久化(六)
相关文章
- PHP编译参数configure配置详解,以及php.ini说明
- 深入PHP:实现MySQL数据库更新(php更新mysql)
- Linux切换PHP版本:快速而安全的方法(linux切换php版本)
- 的应用利用PHP中MySQL函数实现功能强大的数据库操作(php中mysql函数)
- 数据库解决PHP无法连接MySQL数据库的问题.(php无法连接mysql)
- PHP使用Redis实现高性能缓存(php-redis)
- php在Linux上使用Yaf框架开发PHP应用(yaflinux)
- Linux搭建PHP网站实现快速上线(linux架设php网站)
- 数据库MySQL 数据库的 PHP 扩展研究(php扩展mysql)
- 迈出新步伐:Linux下快速PHP升级(linux下php升级)
- 实现PHP程序Redis连接池的简单方法(redis连接池php)
- PHP实现Redis的访问控制(php访问redis)
- 编程开发之路:PHP 与 MySQL 结缘(php与mysql程序设计)
- PHP连接MySQL数据库的实现代码分享(php连接mysql代码)
- PHP开启Redis之路 正确的开启方式(php开启redis)
- Linux系统如何实现PHP开机自启动?(linux开机启动php)
- 深入探究:MySQL与PHP的关联及优化技巧(mysql关联php)
- MySQL PHP长连接:持久化提升应用性能(mysql php长连接)
- PHP结合MySQL实现数据库事务锁定机制(php mysql 锁)
- 用PHP访问本地MSSQL数据库的实现方法(php访问本地mssql)
- PHP环境搭建:从编译MSSQL说起(php 编译mssql)
- PHP实现从MSSQL中数据显示(php 显示 mssql)
- PHP与MySQL实现无缝连接(php mysql 连接)
- PHP实现稳定支持Redis的精彩瞬间(让php支持redis)
- 利用PHP实现Redis队列任务的并发处理(redis队列任务php)
- 使用Redis锁实现PHP程序的安全性(redis锁机制 php)
- Oracle和PHP完美配合提升数据库性能(oracle php配置)
- 使用PHP实现Redis自旋锁(redis自旋锁php)
- 用PHP实现Redis缓存一个实例(redis缓存实例php)
- 中的应用使用Redis,提升PHP程序效率(redis缓存在php)
- PHP安全技术之实现php基本安全
- 微盾PHP脚本加密专家php解密算法
- php数据结构与算法(PHP描述)查找与二分法查找
- PHP基于php_imagick_st-Q8.dll实现JPG合成GIF图片的方法
- PHP针对JSON操作实例分析