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安裝步驟：（已安裝可忽略）

yum  install -y libxml2-devel libcurl-devel which

wget -O php.tar.gz https://www.php.net/distributions/php-7.1.29.tar.gz

mkdir /root/php/ /root/php-cpp/ /root/rocketmq-client-php

tar xvf php.tar.gz -C /root/php/ --strip-components=1

cd /root/php

/configure --prefix=/usr/local/php/ && make && make install

ln -s /usr/local/php/bin/php-config /usr/local/bin/

ln -s /usr/local/php/bin/php /usr/local/bin/

ln -s /usr/local/php/bin/phpize /usr/local/bin/

 

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();

　　

搜索

復制