1. windows RabbitMQ 的使用
  2.  
  3. RabbitMQ是一个在AMQP协议的基础上实现的企业级消息系统。
  4.  
  5. 一、安装
  6. 1、下载erlang
  7. http://erlang.org/download/otp_win64_21.1.exe
  8. 直接安装,设置环境变量 C:\Program Files\erl10.1
  9. 2、下载rabbitmq
  10. https://dl.bintray.com/rabbitmq/all/rabbitmq-server/3.7.8/rabbitmq-server-3.7.8.exe
  11. 直接安装
  12. 3.配置
  13. 激活 RabbitMQ's Management Plugin
  14. 输入命令:
  15. "C:\Program Files\RabbitMQ Server\rabbitmq_server-3.7.8\sbin\rabbitmq-plugins.bat" enable rabbitmq_management
  16. 重启服务:
  17. net stop RabbitMQ && net start RabbitMQ
  18.  
  19. 二、操作
  20. 1、用户角色管理
  21. 进入目录:C:\Program Files\RabbitMQ Server\rabbitmq_server-3.7.8\sbin
  22. 显示用户列表:rabbitmqctl.bat list_users
  23. 新增一个用户:rabbitmqctl.bat add_user raoxiaoya raoxiaoya
  24. 修改用户角色:rabbitmqctl.bat set_user_tags raoxiaoya administrator
  25. 修改用户密码:rabbitmqctl change_password guest guest
  26.  
  27. 2、可视化管理
  28. http:://localhost:15672
  29. 输入用户名密码登录
  30.  
  31. 三、基本概念
  32. 按照运行流程分为以下四个部分:
  33. 1、producer:消息生产者,就是投递消息的程序。
  34. 2、Broker:简单来说就是消息队列服务器实体。
  35. vhost:虚拟主机,一个broker里可以开设多个vhost,用作不同用户的而独立运行。
  36.   Exchange:消息交换机,它指定消息按什么规则,路由到哪个队列。
  37.   Queue:消息队列载体,每个消息都会被投入到一个或多个队列。
  38.   Binding:绑定,它的作用就是把exchange和queue按照路由规则绑定起来。
  39.   Routing Key:路由关键字,exchange根据这个关键字进行消息投递。
  40.   3、connetion:客户端和Broker之间的TCP连接
  41. channel:信道,仅仅创建了客户端到Broker之间的连接后,客户端还是不能发送消息的。需要为每一个Connection创建Channel,AMQP协议规定只有通过Channel才能执行AMQP的命令。一个Connection可以包含多个Channel。之所以需要Channel,是因为TCP连接的建立和释放都是十分昂贵的,如果一个客户端每一个线程都需要与Broker交互,如果每一个线程都建立一个TCP连接,暂且不考虑TCP连接是否浪费,就算操作系统也无法承受每秒建立如此多的TCP连接,可以简单的理解为线程池中的一个个线程。
  42. 4、consumer:消息消费者,就是接受消息的程序。
  43.  
  44. binding:
  45. Binding联系了Exchange与Message Queue。Exchange在与多个Message Queue发生Binding后会生成一张路由表,路由表中存储着Message Queue所需消息的限制条件即Binding Key。当Exchange收到Message时会解析其Header得到Routing Key,Exchange根据Routing Key与Exchange Type将Message路由到Message Queue。
  46. Binding Key由Consumer在Binding Exchange与Message Queue时指定。
  47. Routing Key由Producer发送Message时指定,两者的匹配方式由Exchange Type决定,就好比于邮件上面的地址。
  48. 注意:
  49. 1、生产者是通过交换机来发消息的
  50. 2、除default Exchange之外,如果交换机没有被binding到队列,那么消息就丢失了
  51. 3、Direct Exchange,Topic Exchange进行Binding的时候,需要指定Routing key
  52. 4、Fanout Exchange,Headers Exchange进行Binding的时候,不需要指定Routing key。
  53. 5、消费者第一次binding了就会被保存下来,即使关闭了该消费者,消息依然可以顺利的由交换机到达队列中,再次开启消费者即可消费
  54. 6、为什么是由消费者来binding???
  55. 交换机的类型:
  56. 1、Direct 直接交换器,工作方式类似于单播,Exchange会将消息发送完全匹配ROUTING_KEY的Queue。
  57. 2、fanout广播是式交换器,不管消息的ROUTING_KEY设置为什么,Exchange都会将消息转发给所有绑定的Queue。
  58. 3、topic主题交换器,工作方式类似于组播,Exchange会将消息转发和ROUTING_KEY匹配模式相同的所有队列,比如,ROUTING_KEY为user.stock的Message会转发给绑定匹配模式为 * .stock,user.stock, * . * 和#.user.stock.#的队列。( * 表是匹配一个任意词组,#表示匹配0个或多个词组)。
  59. 4、headers消息体的header匹配(ignore)。
  60.  
  61. 消息队列的使用过程大概如下:
  62. (1)客户端连接到消息队列服务器,打开一个channel。
  63.   (2)客户端声明一个exchange,并设置相关属性。
  64.   (3)客户端声明一个queue,并设置相关属性。
  65.   (4)客户端使用routing key,在exchange和queue之间建立好绑定关系。
  66.   (5)客户端投递消息到exchange。
  67. (6)exchange根据routing key发送到一个或多个队列。
  68. (7)队列将消息发送给消费者。
  69. (8)消费者收到消息,发送ACK给队列确认收到消息。
  70. (9)队列收到ACK,删除队列中缓存的此条消息。
  71.  
  72. 一个交换机对应多个队列实现一条消息同时被多个消费者消费。
  73. 如果一个消费者消费失败,则其指向的队列将不会删除此消息,其他队列不会受影响。
  74. 安全起见,建议做一些辅助的措施。
  75. 一个队列被多个消费者使用实现负载均衡的效果。
  76. 如果cosumer接受了消息, 但在发送ack之前断开连接,rabbitmq会认为这条消息没有被deliver,在consumer在次连接的时候,这条消息会被redeliver。
  77. 如果consumer接受了消息,但是程序中有bug,忘记了ack,rabbitmq不会重复发送消息。
  78.  
  79. RabbitMQ支持消息的持久化,也就是数据写在磁盘上,为了数据安全考虑,我想大多数用户都会选择持久化。消息队列持久化包括3个部分:
  80.   (1)exchange持久化,在声明时指定durable => 1
  81.   (2)queue持久化,在声明时指定durable => 1
  82.   (3)消息持久化,在投递时指定delivery_mode => 2(1是非持久化)
  83. 如果exchange和queue都是持久化的,那么它们之间的binding也是持久化的。如果exchange和queue两者之间有一个持久化,一个非持久化,就不允许建立绑定。
  84.  
  85.  
  86.  
  87. 四、安装php的amqp扩展
  88. 先用phpinfo()查看php版本信息
  89. Compiler
  90. PHP Extension Build
  91.  
  92. http://pecl.php.net/package/amqp
  93. 选择最新版的DLL
  94. 根据php版本,Compiler ,PHP Extension Build 来选择下载
  95.  
  96. 加压后
  97.  
  98. 将php_amqp.dll复制到php/ext,同时在php.ini中添加如下代码:
  99. [amqp]
  100. extension=php_amqp.dll
  101.  
  102. 然后将rabbitmq.4.dll复制到php根目录 D:\phpStudy\PHPTutorial\php\php-7.0.12-nts\
  103.  
  104. 同时修改apache配置文件httpd.conf,添加如下代码:
  105. # rabbitmq
  106. LoadFile "D:/phpStudy/PHPTutorial/php/php-7.0.12-nts/rabbitmq.4.dll"
  107.  
  108. 重启
  109. phpinfo()查看是否加载了 amqp模块
  110.  
  111. 五、php调用RabbitMQ
  112. 消费者
  113. consumer.php
  114.  
  115. <?php
  116. //配置信息
  117. $conn_args = array(
  118. 'host' => '127.0.0.1',
  119. 'port' => '5672',
  120. 'login' => 'guest',
  121. 'password' => 'guest',
  122. 'vhost'=>'/'
  123. );
  124. $e_name = 'e_linvo'; //交换机名
  125. $q_name = 'q_linvo'; //队列名
  126. $k_route = 'key_1'; //路由key
  127. //创建连接和channel
  128. $conn = new AMQPConnection($conn_args);
  129. if (!$conn->connect()) {
  130. die("Cannot connect to the broker!\n");
  131. }
  132. $channel = new AMQPChannel($conn);
  133. //创建交换机
  134. $ex = new AMQPExchange($channel);
  135. $ex->setName($e_name);
  136. $ex->setType(AMQP_EX_TYPE_DIRECT); //direct类型
  137. $ex->setFlags(AMQP_DURABLE); //持久化
  138. echo "Exchange Status:".$ex->declare()."\n";
  139. //创建队列
  140. $q = new AMQPQueue($channel);
  141. $q->setName($q_name);
  142. $q->setFlags(AMQP_DURABLE); //持久化
  143. echo "Message Total:".$q->declare()."\n";
  144. //绑定交换机与队列,并指定路由键
  145. echo 'Queue Bind: '.$q->bind($e_name, $k_route)."\n";
  146. //阻塞模式接收消息
  147. echo "Message:\n";
  148. while(True){
  149. $q->consume('processMessage');
  150. //$q->consume('processMessage', AMQP_AUTOACK); //自动ACK应答
  151. }
  152. $conn->disconnect();
  153. /**
  154. * 消费回调函数
  155. * 处理消息
  156. */
  157. function processMessage($envelope, $queue) {
  158. $msg = $envelope->getBody();
  159. echo $msg."\n"; //处理消息
  160. $queue->ack($envelope->getDeliveryTag()); //手动发送ACK应答
  161. }
  162. ?>
  163.  
  164. 生产者
  165. publish.php
  166.  
  167. <?php
  168. //配置信息
  169. $conn_args = array(
  170. 'host' => '127.0.0.1',
  171. 'port' => '5672',
  172. 'login' => 'guest',
  173. 'password' => 'guest',
  174. 'vhost'=>'/'
  175. );
  176. $e_name = 'e_linvo'; //交换机名
  177. //$q_name = 'q_linvo'; //无需队列名
  178. $k_route = 'key_1'; //路由key
  179. //创建连接和channel
  180. $conn = new AMQPConnection($conn_args);
  181. if (!$conn->connect()) {
  182. die("Cannot connect to the broker!\n");
  183. }
  184. $channel = new AMQPChannel($conn);
  185.  
  186. //创建交换机对象
  187. $ex = new AMQPExchange($channel);
  188. $ex->setName($e_name);
  189. date_default_timezone_set("Asia/Shanghai");
  190. //发送消息
  191. //$channel->startTransaction(); //开始事务
  192. for($i=0; $i<5; ++$i){
  193. sleep(1);//休眠1秒
  194. //消息内容
  195. $message = "TEST MESSAGE!".date("h:i:sa");
  196. echo "Send Message:".$ex->publish($message, $k_route)."\n";
  197. }
  198. //$channel->commitTransaction(); //提交事务
  199. $conn->disconnect();
  200. ?>
  201.  
  202. 分别在两个窗口运行着两个文件
  203.  
  204. 查看网页监控:http:://localhost:15672
  205.  
  206.  
  207. 游戏记录
  208. 服务端:publisher
  209. 交换机:ex_game_record
  210. 路由键:key_game_record
  211. 消息:json 类型
  212.  
  213. 队列1:生成记录
  214. 名称:queue_game_record_log
  215. 交换机:ex_game_record
  216. 路由:key_game_record
  217. 持久化
  218.  
  219. 队列1:复制记录
  220. 名称:queue_game_record_log
  221. 交换机:ex_game_record
  222. 路由:key_game_record
  223. 持久化
  224.  
  225. 队列1:战绩记录
  226. 名称:queue_game_record_log
  227. 交换机:ex_game_record
  228. 路由:key_game_record
  229. 持久化
  230.  
  231. 队列1:统计游戏时间堆叠
  232. 名称:queue_game_record_log
  233. 交换机:ex_game_record
  234. 路由:key_game_record
  235. 持久化

相关评论(0)
您是不是忘了说点什么?

友情提示:垃圾评论一律封号...

还没有评论,快来抢沙发吧!