rocketmq-client-php的安装与使用
业务场景：PHP消费对接Java rocketmq生产
消费模式：Push
处理方案：centos7x搭建rocketmq-client-php，php版本php7x（php -version检查php版本）
参考链接：https://gitee.com/apaas_framework/rocketmq-client-php
操作步骤：
1.安装rocket-client-cpp（基于rocketmq的c++客户端）
yum install -y gcc-c++ cmake automake autoconf libtool bzip2-devel wget tar unzip make zlib-devel
mkdir -p /root/rocketmq-cpp/
wget -O rocketmq.tar.gz https://github.com/apache/rocketmq-client-cpp/archive/1.2.2.tar.gz
tar xvf rocketmq.tar.gz -C /root/rocketmq-cpp/ –strip-components=1
cd /root/rocketmq-cpp/
(安装过程中会下载libevent 2.0.22, jsoncpp 0.10.7, boost 1.58.0，可以提前下载好到rocketmq-cpp目录下方便安装)
sh build.sh
yum clean all
mkdir -p /usr/include/rocketmq
cp /root/rocketmq-cpp/bin/librocketmq.* /usr/lib64/ -rf
cp /root/rocketmq-cpp/include/* /usr/include/rocketmq/ -rf
cp /root/rocketmq-cpp/bin/include/* /usr/include/ -rf
cp /root/rocketmq-cpp/bin/lib/* /usr/lib64/ -rf
rm -rf /root/rocketmq*
2.PHP7安装步骤：（已安装可忽略）
3.安装php-cpp（C++库,用来开发PHP的扩展）
wget -O php-cpp.tar.gz https://github.com/CopernicaMarketingSoftware/PHP-CPP/archive/v2.1.4.tar.gz
tar xvf php-cpp.tar.gz -C /root/php-cpp –strip-components=1
cd /root/php-cpp && make && make install
4.下载rocketmq-client-php并编译安装
wget -O /root/rocketmq-client-php.tar.gz https://github.com/lpflpf/rocketmq-client-php/archive/v0.1-beta.tar.gz
tar xvf /root/rocketmq-client-php.tar.gz -C /root/rocketmq-client-php –strip-components=1
cd /root/rocketmq-client-php/
make && make install
编译成功后会自动将扩展包rocketmq.so添加到php的扩展文件夹中，检查是否存在，不存在复制so文件到扩展文件夹
在php.ini文件中添加extension=rocketmq.so
重启PHP服务，可通过phpinfo查看扩展是否添加成功
用法：示例在rocket-client-php的example目录
我是用PushConsumer连接mq，经测试可以正常使用；其他两个示例没有验证
我的代码：
<?php namespace RocketMQ; include("Message.php"); do{ $consumer = new PushConsumer("test"); $consumer->setInstanceName("test"); $consumer->setNamesrvAddr("127.0.0.1:9876"); $consumer->setThreadCount(1); //单线程 $consumer->setMaxRequestTime(3600); $consumer->setListenerType(MessageListenerType::LISTENER_ORDERLY); $result = array(); $count = 0; // if thread > 1 & use echo method will core dump. $consumer->setCallback(function ($msg) use (&$count){ echo_msg($msg); $count ++; $message = $msg->getMessage(); $body = $message->getBody();
//todo::处理报文 }); $consumer->subscribe("your topic name", "*"); $consumer->start(); $consumer->shutdown(); sleep(10); }while(1);
使用nohup命令在运行该程序即可，nohup php PushConsumer.php &
Producer：
namespace RocketMQ;
$instanceName = "MessageQueue";
$producer = new Producer($instanceName);
$producer->setInstanceName($instanceName);
$producer->setNamesrvAddr("127.0.0.1:9876");
$producer->start();
for ($i = 0; $i < 10000; $i ++){
$message = new Message("TopicTest", "*", "hello world $i");
$sendResult = $producer->send($message);
echo $sendResult->getSendStatus() . "\n";
}
PullConsumer：
It is a good idea to save offset in local.
namespace RocketMQ;
$consumer = new PullConsumer("pullTestGroup");
$consumer->setInstanceName("testGroup");
$consumer->setTopic("TopicTest");
$consumer->setNamesrvAddr("127.0.0.1:9876");
$consumer->start();
$queues = $consumer->getQueues();
foreach($queues as $queue){
$newMsg = true;
$offset = 0;
while($newMsg){
$pullResult = $consumer->pull($queue, "*", $offset, 8);
switch ($pullResult->getPullStatus()){
case PullStatus::FOUND:
foreach($pullResult as $key => $val){
echo $val->getMessage()->getBody() . "\n";
}
$offset += count($pullResult);
break;
default:
$newMsg = false;
break;
}
}
}
PushConsumer:
namespace RocketMQ;
$consumer = new PushConsumer("testGroup");
$consumer->setInstanceName("testGroup");
$consumer->setNamesrvAddr("127.0.0.1:9876");
$consumer->setThreadCount(10);
$consumer->setListenerType(MessageListenerType::LISTENER_ORDERLY);
$count = 0;
$consumer->setCallback(function ($msg) use (&$count){
echo $msg->getMessage()->getBody() . "\n";
$count ++;
});
$consumer->subscribe("TopicTest", "*");
$consumer->start();
$consumer->shutdown();
搜索
复制