PHP AMQP 扩展 应用

feiyang10086

PHP AMQP 扩展 应用

高级消息队列协议（AMQP）是一个异步消息传递所使用的应用层协议规范。作为线路层协议，而不是API（例如JMS），AMQP 客户端能够无视消息的来源任意发送和接受信息。现在，已经有相当一部分不同平台的服务器和客户端可以投入使用

我的AMQP服务器是使用RabbitMQ ，RabbitMQ 的安装网上很多。

重点说下 PHP 的扩展 php-amqp，我用的是最新的 amqp-1.0.1

文档是旧的 新的掺杂在一起 因此弄了好久才完成。

首先说下流程 即 Client - AMQP server - Client

左边的Client向右边的Client发送消息，流程：

1，  获取Conection

2，  获取Channel

3，  定义Exchange，Queue

4，  使用一个RoutingKey将Queue Binding到一个Exchange上

5，  通过指定一个Exchange和一个RoutingKey来将消息发送到对应的Queue上，

6，  接收方在接收时也是获取connection，接着获取channel，然后指定一个Queue直接到它关心的Queue上取消息，它对Exchange，RoutingKey及如何binding都不关心，到对应的Queue上去取消息就OK了

以下是PHP 的实现：

生产消息：

Php代码

1. 1.<?php   
   
2. 2.  
   
3. 3.//设置你的连接   
   
4. 4.$conn_args = array('host' => 'localhost', 'port' => '5672', 'login' => 'guest', 'password' => 'guest');   
   
5. 5.$conn = new AMQPConnection($conn_args);   
   
6. 6.if ($conn->connect()) {   
   
7. 7.    echo "Established a connection to the broker \n";   
   
8. 8.}   
   
9. 9.else {   
   
10. 10.    echo "Cannot connect to the broker \n ";   
    
11. 11.}   
    
12. 12.//你的消息   
    
13. 13.$message = json_encode(array('Hello World!','php','c++'));   
    
14. 14.//创建channel   
    
15. 15.$channel = new AMQPChannel($conn);   
    
16. 16.//创建exchange   
    
17. 17.$ex = new AMQPExchange($channel);   
    
18. 18.$ex->setName('exchange');//创建名字   
    
19. 19.$ex->setType(AMQP_EX_TYPE_DIRECT);   
    
20. 20.$ex->setFlags(AMQP_DURABLE | AMQP_AUTODELETE);   
    
21. 21.echo "exchange status:".$ex->declare();   
    
22. 22.echo "\n";   
    
23. 23.//创建队列   
    
24. 24.$q = new AMQPQueue($channel);   
    
25. 25.//设置队列名字 如果不存在则添加   
    
26. 26.$q->setName('queue');   
    
27. 27.$q->setFlags(AMQP_DURABLE | AMQP_AUTODELETE);   
    
28. 28.echo "queue status: ".$q->declare();   
    
29. 29.echo "\n";   
    
30. 30.echo 'queue bind: '.$q->bind('exchange','route.key');//将你的队列绑定到routingKey   
    
31. 31.echo "\n";   
    
32. 32.  
    
33. 33.$channel->startTransaction();   
    
34. 34.echo "send: ".$ex->publish($message, 'route.key'); //将你的消息通过制定routingKey发送   
    
35. 35.$channel->commitTransaction();   
    
36. 36.$conn->disconnect();   
    
37. 37.  
    
38. 38.?>  
    
39. <?php
    
40. 
    
41. //设置你的连接
    
42. $conn_args = array('host' => 'localhost', 'port' => '5672', 'login' => 'guest', 'password' => 'guest');
    
43. $conn = new AMQPConnection($conn_args);
    
44. if ($conn->connect()) {
    
45.     echo "Established a connection to the broker \n";
    
46. }
    
47. else {
    
48.     echo "Cannot connect to the broker \n ";
    
49. }
    
50. //你的消息
    
51. $message = json_encode(array('Hello World!','php','c++'));
    
52. //创建channel
    
53. $channel = new AMQPChannel($conn);
    
54. //创建exchange
    
55. $ex = new AMQPExchange($channel);
    
56. $ex->setName('exchange');//创建名字
    
57. $ex->setType(AMQP_EX_TYPE_DIRECT);
    
58. $ex->setFlags(AMQP_DURABLE | AMQP_AUTODELETE);
    
59. echo "exchange status:".$ex->declare();
    
60. echo "\n";
    
61. //创建队列
    
62. $q = new AMQPQueue($channel);
    
63. //设置队列名字 如果不存在则添加
    
64. $q->setName('queue');
    
65. $q->setFlags(AMQP_DURABLE | AMQP_AUTODELETE);
    
66. echo "queue status: ".$q->declare();
    
67. echo "\n";
    
68. echo 'queue bind: '.$q->bind('exchange','route.key');//将你的队列绑定到routingKey
    
69. echo "\n";
    
70. 
    
71. $channel->startTransaction();
    
72. echo "send: ".$ex->publish($message, 'route.key'); //将你的消息通过制定routingKey发送
    
73. $channel->commitTransaction();
    
74. $conn->disconnect();
    
75. 
    
76. ?>

复制代码

接收方 消费方

Php代码

1. 1.  
   
2. 2.<?php   
   
3. 3.$conn_args = array('host' => 'localhost', 'port' => '5672', 'login' => 'guest', 'password' => 'guest'  
   
4. 4.                    ,'vhost'=>'/');   
   
5. 5.$conn = new AMQPConnection($conn_args);   
   
6. 6.$conn->connect();   
   
7. 7.$channel = new AMQPChannel($conn);   
   
8. 8.$q = new AMQPQueue($channel);   
   
9. 9.$q->setName('queue2');   
   
10. 10.$q->setFlags(AMQP_DURABLE | AMQP_AUTODELETE);   
    
11. 11.echo "queue status: ".$q->declare();   
    
12. 12.echo "==========\n";   
    
13. 13.        
    
14. 14.$messages = $q->get(AMQP_AUTOACK);   
    
15. 15.print_r($messages->getBody());   
    
16. 16.echo "\n";   
    
17. 17.// disconnect   
    
18. 18.$conn->disconnect();   
    
19. 19.  
    
20. 20.?>  
    
21. 
    
22. <?php
    
23. $conn_args = array('host' => 'localhost', 'port' => '5672', 'login' => 'guest', 'password' => 'guest'
    
24.                     ,'vhost'=>'/');
    
25. $conn = new AMQPConnection($conn_args);
    
26. $conn->connect();
    
27. $channel = new AMQPChannel($conn);
    
28. $q = new AMQPQueue($channel);
    
29. $q->setName('queue2');
    
30. $q->setFlags(AMQP_DURABLE | AMQP_AUTODELETE);
    
31. echo "queue status: ".$q->declare();
    
32. echo "==========\n";
    
33.      
    
34. $messages = $q->get(AMQP_AUTOACK);
    
35. print_r($messages->getBody());
    
36. echo "\n";
    
37. // disconnect
    
38. $conn->disconnect();
    
39. 
    
40. ?>

复制代码

我是初学 消息队列这方面知识。
不知道 linux的消息队列 和AMQP 的 优缺点
我用php 的sysvmsg 函数 执行while 循环 可以挂起在服务器上，实时监听消息队列的消息，有消息则取，内存不会升高

而用amqp while 循环 监听 即使无消息  内存也会越来越高，最后导致PHP 使用内存不足而 报错。

看来这方面得多交流 多学习

如果有一天21

谢谢分享