PHP用ActiveMq 实现消息列队

2023-03-10 19:30:06

1.各种安装

2.简单配置：

jetty.xml

localhost:8161

配置：

activemq添加stomp的61613接口

conf/activemq.xml

<transportConnector name="stomp+nio" uri="stomp+nio://0.0.0.0:61613?transport.closeAsync=false"/>

3.创建消息列队

<?php

$queue = "/queue/userReg";

try {

    $stomp = new Stomp('tcp://127.0.0.1:61613');

} catch (StompException $e) {

    die('Connection failed: ' . $e->getMessage());

}

$obj = new Stdclass();

$obj->username = 'smartom';

$obj->password = '123';

$obj->status = 1;

$stomp->send('/queue/userReg', json_encode($obj));

4.销毁消息列队

<?php

$queue = "/queue/userReg";

try {

    $stomp = new Stomp('tcp://127.0.0.1:61613');

} catch (StompException $e) {

    die('Connection failed: ' . $e->getMessage());

}

$isSubscribe = $stomp->subscribe($queue);       //订阅

while ($stomp->hasFrame()) {                      //循环读取队列

	$frame = $stomp->readFrame();				  //读取下一个

    $user =  json_decode($frame->body);

    /*

    处理消息

    */

    if($user->status=1){

        $user->status=2;

    	$stomp->send('/queue/sendVerify', $frame->body);  //发送下一个

    	$stomp->ack($frame);		//销毁列队

    }

    sleep(3);

}

5.开启事务

$queue1 = "/queue/userReg";

$queue2 = "/queue/sendMsg";

$stomp->begin("news");

  if($stomp->send($queue1,json_encode($mainSite), array('transaction' => 'news')) && $stomp->send($queue2,json_encode($searchIndex) , array('transaction' => 'news')) )//发布到主站消息

 {

        $stomp->commit("news");

 }

6.topic多个订阅者

queue是单个订阅者

$broker = 'tcp://192.168.222.129:61613';

$queue = "/topic/order_create";

7.集群配置

8.共享文件系统

使用共享系统来做 Master/slave集群

9.共享数据库方式

所有消息读写其实都是数据库操作，性能完全取代与数据库的性能。

10.基于zookeeper 的主从(levelDB Master/Slave)

基于zookeeper 来选举出一个master ,其他节点自动作为slave实时同步消息.

安装相应工具

yum -y install nfs-utils rpcbind

nfs(network file system) 网络文件系统

rpcbind 是一个RPC服务，主要是在nfs 共享时负责通知客户端，服务器的nfs端口号

码农公寓