1yum install librdkafka-devel

1pecl install rdkafka
2# 在PHP的初始化文件php.ini中添加以下语句以开启扩展
3extension=rdkafka.so
4php -m | grep kafka
5rdkafka

1$ git clone https://github.com/arnaud-lb/php-rdkafka.git
2$ cd php-rdkafka
3$ phpize
4$ ./configure
5$ make all -j 5
6$ sudo make install

1<?php
2
3$conf = new RdKafka\Conf();
4// 接入点
5$conf->set('metadata.broker.list', '接入点');
6// 接入协议
7$conf->set('security.protocol','ssl');
8// 证书文件路径（证书文件请参考"接入点查看"文档）
9$conf->set('ssl.ca.location',__DIR__.'/ssl.cert/ca.pem');
10$conf->set('ssl.certificate.location',__DIR__.'/ssl.cert/client.pem');
11$conf->set('ssl.key.location ',__DIR__.'/ssl.cert/client.key');
12
13$producer = new RdKafka\Producer($conf);
14
15// 指定topic
16$topic = $producer->newTopic("topic_name");
17
18for ($i = 0; $i < 10; $i++) {
19    // RD_KAFKA_PARTITION_UA自动选择分区，message指定想要发送的消息
20    $topic->produce(RD_KAFKA_PARTITION_UA, 0, "Message $i : php ssl");
21    $producer->poll(0);
22}
23
24for ($flushRetries = 0; $flushRetries < 10; $flushRetries++) {
25    $result = $producer->flush(10000);
26    if (RD_KAFKA_RESP_ERR_NO_ERROR === $result) {
27        break;
28    }
29}
30
31if (RD_KAFKA_RESP_ERR_NO_ERROR !== $result) {
32    throw new \RuntimeException('Was unable to flush, messages might be lost!');
33}

1<?php
2
3$conf = new RdKafka\Conf();
4// 接入协议
5$conf->set('security.protocol','ssl');
6// 证书文件路径
7$conf->set('ssl.ca.location',__DIR__.'/ssl.cert/ca.pem');
8$conf->set('ssl.certificate.location',__DIR__.'/ssl.cert/client.pem');
9$conf->set('ssl.key.location ',__DIR__.'/ssl.cert/client.key');
10// 消费组 id
11$conf->set('group.id', 'php-group');
12
13$rk = new RdKafka\Consumer($conf);
14// 接入点
15$rk->addBrokers("接入点");
16
17$topicConf = new RdKafka\TopicConf();
18$topicConf->set('auto.commit.interval.ms', 100);
19$topicConf->set('offset.store.method', 'broker');
20$topicConf->set('auto.offset.reset', 'earliest');
21
22//订阅topic
23$topic = $rk->newTopic("topic_name", $topicConf);
24
25$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);
26
27while (true) {
28    $message = $topic->consume(0, 120*10000);
29    switch ($message->err) {
30        case RD_KAFKA_RESP_ERR_NO_ERROR:
31            var_dump($message);
32            break;
33        case RD_KAFKA_RESP_ERR__PARTITION_EOF:
34            echo "No more messages; will wait for more\n";
35            break;
36        case RD_KAFKA_RESP_ERR__TIMED_OUT:
37            echo "Timed out\n";
38            break;
39        default:
40            throw new \Exception($message->errstr(), $message->err);
41            break;
42    }
43}

1# 启动消费者
2php consumer.php
3# 启动生产者
4php producer.php