本文中的相关环境:
操作系统版本:CentOS Linux release 7.8.2003 (Core)
php版本:PHP 7.1.33 ( NTS )
Stomp扩展版本: 2.0.1
php在操作activemq的时候需要借助一个叫Stomp的扩展,所以要先安装一下该扩展。PS:准确来说,php的Stomp扩展是基于Stomp协议来开发的,而activemq支持多种协议(其中就包括 Stomp协议),所以 php也是通过Stomp协议(php的Stomp扩展)来和activemq进行交互的
安装php的Stomp扩展(pecl一键安装方式)
pecl install https://pecl.php.net/get/stomp-2.0.1.tgz -y # 安装过程中出现 OpenSSL install prefix (no to disable SSL support) [/usr] : # 这里输入yes 按回车继续安装,不输入yes直接按回车 在接下来的编译中会报错 导致安装失败 # 安装完毕后可以在 cli模式使用 php -m | grep Stomp 来查看扩展是否已安装 或 在web模式下使用 phpinfo() 函数来查看是否已安装该扩展
Stomp扩展的各大版本地址:https://pecl.php.net/package/stomp
创建队列,并往队列中写入数据:
/***************** 队列类型是:queue *************************/ $broker = 'tcp://192.168.13.119:61613'; //61613是默认的端口,在activemq目录下的conf/activemq.xml文件可查看到 $queue = '/queue/userreg'; //创建一个queue类型的队列,队列名字是userreg 注意 这里是queue!这里是queue!这里是queue! $user_data = array('user_name' => 'zhangsan'); $stomp = new Stomp($broker); $stomp->send($queue, json_encode($user_data)); /***************** 队列类型是:topic *************************/ $broker = 'tcp://192.168.13.119:61613'; $topic_order = '/topic/order'; //创建一个topic类型的队列,队列名字是order 注意 这里是topic!这里是topic!这里是topic! $order_data = array('order_sn' => '20201125155856'); $stomp = new Stomp($broker); $stomp->send($topic_order, json_encode($order_data)); //queue和topic队列类型的区别 是在于读取队列的时候,queue用于点对点模型(一个生产者,一个消费者 是1对1的关系), topic用于发布/订阅模型(一个生产者,多个消费者 是1对多的关系)
普通队列读取(点对点(Point To Point)模型),使用示例:
$broker = 'tcp://192.168.13.119:61613'; $queue = '/queue/userreg'; //注意 这里是queue!这里是queue!这里是queue! try { $stomp = new Stomp($broker); $stomp->subscribe($queue); ////订阅了一个queue类型的队列 队列名字是userreg $frame = $stomp->readFrame(); //读取队列 echo $frame->body.PHP_EOL; //输出队列中的具体消息内容 $stomp->ack($frame); //给activemq服务发送ack通知,activemq服务收到ack通知后会把该消息从队列中移除 } catch(StompException $e) { echo $e->getMessage(); } //一个队列中可能会有很多个数据,一般都是循环读取,示例代码如下: $broker = 'tcp://192.168.13.119:61613'; $queue = '/queue/userreg'; //注意 这里是queue!这里是queue!这里是queue! try{ $stomp = new Stomp($broker); $stomp->subscribe($queue); //hasFrame()方法:判断队列中是否有数据, false:队列中没有数据 true:队列中有数据 while($stomp->hasFrame() !== false) { $frame = $stomp->readFrame(); echo $frame->body.PHP_EOL; $stomp->ack($frame); usleep(5000000); //休眠5000毫秒 } }catch(StompException $e) { echo $e->getMessage(); }
点对点消息模型特点:只有一个消费者可以接收到消息,并且不能进行重复消费。
发布/订阅(Publish/Subscribe)模型,使用示例:
$broker = 'tcp://192.168.13.119:61613'; $topic= '/topic/order'; //注意 这里是topic!这里是topic!这里是topic! try{ $stomp = new Stomp($broker); $stomp->subscribe($topic); //订阅了一个topic类型的队列 队列名字是order //hasFrame()方法:判断队列中是否有数据, false:队列中没有数据 true:队列中有数据 while($stomp->hasFrame() !== false) { $frame = $stomp->readFrame(); echo $frame->body.PHP_EOL; $stomp->ack($frame); usleep(5000000); //休眠5000毫秒 } }catch(StompException $e) { echo $e->getMessage(); } //假如说下面还有1、2...N份代码都订阅了 /topic/order 队列,那么只要往/topic/order 发送消息后,1、2...N份代码都会同时读取到 /topic/order 队列中的内容,前提是 1、2...N份代码都在实时监听 /topic/order 队列才能收到
发布/订阅模型特点:消息是1对多的、多个消费者可以收到消息、可以重复消费消息
activemq中的事物处理
$broker = 'tcp://192.168.13.119:61613'; $queue1 = '/queue/sendsms'; $queue2 = '/queue/search'; $stomp = new Stomp($broker); $stomp->subscribe($queue); $stomp->begin("news"); if($stomp->send($queue1, 'sendsms队列中的内容', array('transaction' => 'news')) && $stomp->send($queue2, 'search队列中的内容', array('transaction' => 'news')) ) { $stomp->commit("news"); } //begin方法中的值"news" 可以随便填写,不过下面的transaction的值和commit方法中的值必须和begin中的值是一样的
声明:禁止任何非法用途使用,凡因违规使用而引起的任何法律纠纷,本站概不负责。
精彩评论