php amqp 消息队列 RabbitMQ 交换器类型 直连 (三)
发布时间：2015-05-27 19:22:20作者：知识屋
1、AMQP_EX_TYPE_DIRECT:直连型
直连型又包括： 1对1 和1对N（N对1、 N对N）
接收端receive.php代码如下
connect();$channel = new AMQPChannel($connect);$exchange = new AMQPExchange($channel);$exchange->setName('exchange');$exchange->setType(AMQP_EX_TYPE_DIRECT);$exchange->declare();$queue = new AMQPQueue($channel);$queue->setName('logs');$queue->declare();$queue->bind('exchange', 'logs');while (true) { $queue->consume('callback');}$connection->close();function callback($envelope, $queue) { var_dump($envelope->getBody()); $queue->nack($envelope->getDeliveryTag());}发送端send.php代码如下
connect();$channel = new AMQPChannel($connect);$exchange = new AMQPExchange($channel);$exchange->setName('exchange');$exchange->setType(AMQP_EX_TYPE_DIRECT);$exchange->declare();$exchange->publish('direct type test','logs');var_dump("Send Message OK");$connect->disconnect();运行结果如图所示
创建receive_one.php和receive_two.php 并把send.php代码改成如下代码方便我们观看receive_one.php 和 receive_two.php 代码相同 或者用dos运行多个接收端connect();$channel = new AMQPChannel($connect);$exchange = new AMQPExchange($channel);$exchange->setName('exchange');$exchange->setType(AMQP_EX_TYPE_DIRECT);$exchange->declare();$queue = new AMQPQueue($channel);$queue->setName('logs');@$queue->declare();$queue->bind('exchange', 'logs');while (true) { $queue->consume('callback');}$connection->close();function callback($envelope, $queue) { var_dump($envelope->getBody()); $queue->nack($envelope->getDeliveryTag());}send.php
connect();$channel = new AMQPChannel($connect);$exchange = new AMQPExchange($channel);$exchange->setName('exchange');$exchange->setType(AMQP_EX_TYPE_DIRECT);$exchange->declare();for ($index = 1; $index < 5; $index++) { $exchange->publish($index,'logs'); var_dump("Send:$index");}$exchange->delete();$connect->disconnect();运行结果如下
列队会把消息分配给每一个接收端分配处理这里看似完美但是如果想要更好的处理不同的任务就需要 公平调度
比如当1、3处理的都是简单的人 2、4都是处理的复杂的任务 如果任务过多时 receive_one.php是空闲的而receive_two.php是任务繁重的我们进行如下测试send.php改成5改成50
for ($index = 1; $index < 50; $index++) { $exchange->publish($index,'logs'); var_dump("Send:$index");}receive_two.php 加上 sleep(3)
function callback($envelope, $queue) { var_dump($envelope->getBody()); sleep(3); $queue->nack($envelope->getDeliveryTag());}我们运行程序结果如下
receive_one全部运行完而receive_two才运行一个 之后receive_one一直空闲我们可以通过 在接收端设置$channel->setPrefetchCount(1);
任务没人完成前不接收新的消息把消息发送给其他接收端
如下receive_one.php 和 receive_two.php
$channel = new AMQPChannel($connect);改成如下
$channel = new AMQPChannel($connect);$channel->setPrefetchCount(1);
（免责声明：文章内容如涉及作品内容、版权和其它问题，请及时与我们联系，我们将在第一时间删除内容，文章内容仅供参考）
相关知识
-
如何对PHP文件进行加密方法 PHP实现加密的几种方式介绍
-
php生成圆角图片的方法 电脑中php怎么生成圆角图片教程
-
用PHP构建一个留言本方法步骤 php怎么实现留言板功能
-
php中三元运算符用法 php中的三元运算符使用说明
-
php文件如何怎么打开方式介绍 php文件用什么打开方法
-
PHP怎么插入数据库方法步骤 php编程怎么导入数据库教程
-
如何安装PHPstorm并配置方法教程 phpstorm安装后要进行哪些配置
-
PHP 获取远程文件大小的3种解决方法 如何用PHP获取远程大文件的大小
-
20个实用PHP实例代码 php接口开发实例代码详细介绍
-
如何架设PHP服务器方法步骤 怎么搭建php服务器简单教程
软件推荐
更多 >
-
1
-
2
php MYSQL数据操作类
-
3
PHP单元测试利器：PHPUNIT初探
-
4
php .htaccess 伪静态
-
5
PHP整合PayPal支付
-
6
PHP与ASP比较：这还值得讨论吗？
-
7
PHP与XML结合的网站编程
-
8
Warning: ftp_put() 问题的解决办法之被动模式
-
9
PHP Socket 编程
-
10
PHP性能优化总结
