php 利用activeMq+stomp实现消息队列

php 利用activeMq+stomp实现消息队列

1、activeMq概述

  ActiveMQ 是Apache出品,最流行的,能力强劲的开源消息总线。ActiveMQ 是一个彻底支持JMS1.1和J2EE 1.4规范的 JMS Provider实现,尽管JMS规范出台已是好久的事情了,可是JMS在当今的J2EE应用中间仍然扮演着特殊的地位。php

2、特性列表

⒈ 多种语言和协议编写客户端。语言: Java,C,C++,C#, Ruby, Perl, Python, PHP。应用协议: OpenWire,Stomp REST,WS Notification,XMPP, AMQP
⒉ 彻底支持 JMS1.1和 J2EE 1.4规范 (持久化,XA消息, 事务)
⒊ 对Spring的支持,ActiveMQ能够很容易内嵌到使用Spring的系统里面去,并且也支持Spring2.0的特性
⒋ 经过了常见J2EE 服务器(如 Geronimo,JBoss 4,GlassFish,WebLogic)的测试,其中经过JCA 1.5 resource adaptors的配置,可让ActiveMQ能够自动的部署到任何兼容J2EE 1.4 商业服务器上
⒌ 支持多种传送协议:in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA
⒍ 支持经过JDBC和journal提供高速的消息持久化
⒎ 从设计上保证了高性能的集群,客户端- 服务器,点对点
⒏ 支持 Ajax
⒐ 支持与Axis的整合
⒑ 能够很容易的调用内嵌JMS provider,进行测试

3、运行环境

  在本地或者服务器中安装activeMq,以Windows为例:java

A、 windows下部署apache

ActiveMQ部署其实很简单,和全部Java同样,要跑java程序就必须先安装JDK并配置好环境变量,这个很简单。json

而后解压下载的apache-activemq-5.10-20140603.133406-78-bin.zip压缩包到一个目录,获得解压后的目录结构以下图:windows

进入bin目录,发现有win32和win64两个文件夹,这2个文件夹分别对应windows32位和windows64位操做系统的启动脚本。浏览器

个人实验环境是windowsXP,就进入win32目录,会看到以下目录结构。服务器

其中activemq.bat即是启动脚本,双击启动。app

ActiveMQ默认启动到8161端口,启动完了后在浏览器地址栏输入:http://localhost:8161/admin要求输入用户名密码,默认用户名密码为admin、admin,这个用户名密码是在conf/users.properties中配置的。输入用户名密码后即可看到以下图的ActiveMQ控制台界面了。tcp

4、生产消息

  本文使用stomp协议实现mq队列 ,项目中要加载stomp 包;  ide

  在须要使用队列的地方,导入该包:

  

use App\Libraries\Stomp\Stomp;

  将消息放入activeMq中:

  

    protected function sendToMQ($destination, $msg_data, $persistent = false)
    {
        try {
            $con = new Stomp(config('app.mq_url'));
            $con->connect();
            $con->begin("Transaction");
            $con->send($destination, json_encode($msg_data), array('persistent'=> $persistent));
            $con->commit("Transaction");
            $con->disconnect();
        } catch (\Exception $e) {
            app('log')->warn($e->getMessage());
        }
    }

  1.destination是指队列名称;

  2.msg_data队列中存放的数据;

  3.persistent是否同步; 

  4.app.mq_url是指activeMq安装的服务器地址及端口号,如tcp://localhost:61613;

  通过段代码以后,咱们就将数据msg_data放到了队列destination中了;

5、消费消息

  在activeMq中的消息如何消费呢?通常状况下咱们会创建一个cronjob来定时消费队列的消息,消费队列消息主要代码以下:

        try {
            $this->consumer = new Stomp($this->activemq_uri); //$this->acitvemq_uri就是上面的activeMq地址app.mq_url
            if (!$this->consumer->isConnected()) {
                $this->consumer->connect();
                $this->consumer->setReadTimeout(3);
            }
            app('log')->info($this->log_remark . "connect to active mq success");
        } catch (StompException $e) {
            app('log')->info("connect to active mq failed : " . $e->getMessage());
            die();
        }
        $queue = self::getQueue();//获得队列名称,上面定义的destination
        $this->consumer->subscribe($queue);  //订阅
        //循环读帧
        while ($this->consumer->hasFrameToRead()) {
            try {
                $message = $this->consumer->readFrame();
                $data = json_decode($message->body, true);  // 这里其实就是获得了上面的队列消息msg_data
                //验证数据并更新数据
                $handle_result = self::removeDuplicateExpressInfo($data['traces'], $data['shipping_id']);
                if ($handle_result) {
                    $this->consumer->ack($message);
                }
            } catch (\Exception $e) {
                app('log')->info("handle with message failed : " . $e->getMessage());
                if (!$this->consumer->isConnected()) {
                    $this->consumer->connect();
                } else {
                    break;
                }
            }
        }
        $this->consumer->unsubscribe($queue);    //释放订阅
        $this->consumer->disconnect();      //端口链接
相关文章
相关标签/搜索