本文介紹如何使用PHP SDK通過接入點接入云消息隊列 Kafka 版并收發消息。
環境準備
安裝C++依賴庫
執行以下命令切換到yum源配置目錄/etc/yum.repos.d/。
cd /etc/yum.repos.d/
創建yum源配置文件confluent.repo。
[Confluent.dist] name=Confluent repository (dist) baseurl=https://packages.confluent.io/rpm/5.1/7 gpgcheck=1 gpgkey=https://packages.confluent.io/rpm/5.1/archive.key enabled=1 [Confluent] name=Confluent repository baseurl=https://packages.confluent.io/rpm/5.1 gpgcheck=1 gpgkey=https://packages.confluent.io/rpm/5.1/archive.key enabled=1
執行以下命令安裝C++依賴庫。
yum install librdkafka-devel
安裝PHP依賴庫
執行以下命令安裝PHP依賴庫。
pecl install rdkafka
在PHP的初始化文件php.ini中添加以下一行語句以開啟擴展。
extension=rdkafka.so
準備配置
可選:下載SSL根證書。如果是SSL接入點,需下載該證書。
訪問Aliware-kafka-demos,單擊,下載Demo工程到本地并解壓。
在解壓的Demo工程找到kafka-php-demo文件夾,根據接入點類型打開對應的文件夾,配置setting.php文件。
<?php return [ 'sasl_plain_username' => 'xxx', 'sasl_plain_password' => 'xxx', 'bootstrap_servers' => "xxx:xx,xxx:xx", 'topic_name' => 'xxx', 'consumer_id' => 'xxx' ];
參數
描述
sasl_plain_username
SASL用戶名。如果是默認接入點,則無此配置項。
說明- 如果實例未開啟ACL,您可以在云消息隊列 Kafka 版控制臺的實例詳情頁面的配置信息區域獲取默認的用戶名和密碼。
- 如果實例已開啟ACL,請確保要使用的SASL用戶已被授予向云消息隊列 Kafka 版實例收發消息的權限。具體操作,請參見SASL用戶授權。
sasl_plain_password
SASL用戶名密碼。如果是默認接入點,則無此配置項。
bootstrap_servers
SSL接入點。您可在云消息隊列 Kafka 版控制臺的實例詳情頁面的接入點信息區域獲取。
topic_name
Topic名稱。您可在云消息隊列 Kafka 版控制臺的Topic 管理頁面獲取。
consumer_id
Group名稱。您可在云消息隊列 Kafka 版控制臺的Group 管理頁面獲取。
配置完成后,將配置文件所在文件夾下的全部文件(如果是SSL接入點實例,包含證書SSL根證書文件)上傳至服務器PHP安裝目錄下。
發送消息
執行以下命令發送消息。
php kafka-producer.php
消息程序kafka-producer.php示例代碼如下:
示例代碼為SSL接入點的代碼。如果是默認接入點,無SASL相關代碼,即刪除如下代碼中包含sasl.
和ssl.
相關的代碼即可。
<?php
$setting = require __DIR__ . '/setting.php';
$conf = new RdKafka\Conf();
$conf->set('sasl.mechanisms', 'PLAIN');
$conf->set('api.version.request', 'true');
$conf->set('sasl.username', $setting['sasl_plain_username']);
$conf->set('sasl.password', $setting['sasl_plain_password']);
$conf->set('security.protocol', 'SASL_SSL');
$conf->set('ssl.ca.location', __DIR__ . '/ca-cert.pem');
$conf->set('ssl.endpoint.identification.algorithm', 'none');
$conf->set('message.send.max.retries', 5);
$rk = new RdKafka\Producer($conf);
# if want to debug, set log level to LOG_DEBUG
$rk->setLogLevel(LOG_INFO);
$rk->addBrokers($setting['bootstrap_servers']);
$topic = $rk->newTopic($setting['topic_name']);
$a = $topic->produce(RD_KAFKA_PARTITION_UA, 0, "Message hello kafka");
$rk->poll(0);
while ($rk->getOutQLen() > 0) {
$rk->poll(50);
}
echo "send succ" . PHP_EOL;
代碼示例詳情,請參見php-rdkafka。
訂閱消息
執行如下命令訂閱消息。
php kafka-consumer.php
消息程序kafka-consumer.php示例代碼如下:
示例代碼為SSL接入點的代碼。如果是默認接入點,無SASL相關代碼,即刪除如下代碼中包含sasl.
和ssl.
相關的代碼即可。
<?php
$setting = require __DIR__ . '/setting.php';
$conf = new RdKafka\Conf();
$conf->set('sasl.mechanisms', 'PLAIN');
$conf->set('api.version.request', 'true');
$conf->set('sasl.username', $setting['sasl_plain_username']);
$conf->set('sasl.password', $setting['sasl_plain_password']);
$conf->set('security.protocol', 'SASL_SSL');
$conf->set('ssl.ca.location', __DIR__ . '/ca-cert.pem');
$conf->set('session.timeout.ms', 10000);
$conf->set('request.timeout.ms', 305000);
$conf->set('group.id', $setting['consumer_id']);
$conf->set('ssl.endpoint.identification.algorithm', 'none');
$conf->set('metadata.broker.list', $setting['bootstrap_servers']);
$topicConf = new RdKafka\TopicConf();
$conf->setDefaultTopicConf($topicConf);
$consumer = new RdKafka\KafkaConsumer($conf);
$consumer->subscribe([$setting['topic_name']]);
echo "Waiting for partition assignment... (make take some time when\n";
echo "quickly re-joining the group after leaving it.)\n";
while (true) {
$message = $consumer->consume(30 * 1000);
switch ($message->err) {
case RD_KAFKA_RESP_ERR_NO_ERROR:
var_dump($message);
break;
case RD_KAFKA_RESP_ERR__PARTITION_EOF:
echo "No more messages; will wait for more\n";
break;
case RD_KAFKA_RESP_ERR__TIMED_OUT:
echo "Timed out\n";
break;
default:
throw new \Exception($message->errstr(), $message->err);
break;
}
}
?>
代碼示例詳情,請參見php-rdkafka。