VPC网络SASL_PLAINTEXT方式生产和消费消息
更新时间：2025-08-21
在同 VPC 网络下访问，使用 SASL_PLAINTEXT 协议接入，接入点可以在 集群详情 页面查看。
环境准备
- 安装C++依赖库
Shell
1yum install librdkafka-devel- 安装PHP依赖库
Shell
1pecl install rdkafka
2# 在PHP的初始化文件php.ini中添加以下语句以开启扩展
3extension=rdkafka.so
4php -m | grep kafka
5rdkafka- 使用php-rdkafka sdk：https://arnaud.le-blanc.net/php-rdkafka-doc/phpdoc/index.html
Shell
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. 购买专享版消息服务for Kafka集群
开通消息服务 for Kafka服务后，在控制台页面点击『创建集群』，即可进行购买。
2. 为购买的集群创建主题
在控制台页面点击集群名称，进入集群详情页面。
在左侧的边栏中点击『主题管理』，进入主题管理页面。
在主题管理页面点击『创建主题』，进行主题的创建。
使用步骤：
步骤一：获取集群接入点
具体请参考：接入点查看。
步骤二：编写测试代码
- 需要关注并自行修改的参数
- 认证机制支持PLAIN、SCRAM-SHA-512两种机制，根据集群所使用的认证方式进行选择
| 参数名 | 含义 |
|---|---|
| metadata.broker.list | 接入点信息 |
| sasl.username | 用户管理中创建用户的用户名 |
| sasl.password | 用户管理中创建用户的密码 |
| topic_name | 主题名称 |
| group_id | 消费组id |
生产者代码示例
创建KafkaProducerDemo.php文件，具体代码示例如下：
Shell
1<?php
2
3$conf = new RdKafka\Conf();
4// 接入点
5$conf->set('metadata.broker.list', '接入点');
6// 接入协议
7$conf->set('security.protocol','sasl_plaintext');
8// SASL 机制
9$conf->set('sasl.mechanism','SCRAM-SHA-512');
10// SASL 用户名
11$conf->set('sasl.username','alice');
12// SASL 密码
13$conf->set('sasl.password','alice1234!');
14
15$producer = new RdKafka\Producer($conf);
16
17// 指定topic
18$topic = $producer->newTopic("topic_name");
19
20for ($i = 0; $i < 10; $i++) {
21 // RD_KAFKA_PARTITION_UA自动选择分区,message指定想要发送的消息
22 $topic->produce(RD_KAFKA_PARTITION_UA, 0, "Message $i : php sasl_plaintext");
23 $producer->poll(0);
24}
25
26for ($flushRetries = 0; $flushRetries < 10; $flushRetries++) {
27 $result = $producer->flush(10000);
28 if (RD_KAFKA_RESP_ERR_NO_ERROR === $result) {
29 break;
30 }
31}
32
33if (RD_KAFKA_RESP_ERR_NO_ERROR !== $result) {
34 throw new \RuntimeException('Was unable to flush, messages might be lost!');
35}
36?>消费者代码示例
创建KafkaConsumerDemo.php文件，具体代码示例如下：
Shell
1<?php
2
3$conf = new RdKafka\Conf();
4// 接入协议
5$conf->set('security.protocol','sasl_plaintext');
6// SASL 机制
7$conf->set('sasl.mechanism','SCRAM-SHA-512');
8// SASL 用户名
9$conf->set('sasl.username','alice');
10// SASL 密码
11$conf->set('sasl.password','alice1234!');
12// 消费组 id
13$conf->set('group.id', 'php-group');
14
15$rk = new RdKafka\Consumer($conf);
16// 接入点
17$rk->addBrokers("接入点");
18
19$topicConf = new RdKafka\TopicConf();
20$topicConf->set('auto.commit.interval.ms', 100);
21$topicConf->set('offset.store.method', 'broker');
22$topicConf->set('auto.offset.reset', 'earliest');
23
24//订阅topic
25$topic = $rk->newTopic("topic_name", $topicConf);
26
27$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);
28
29while (true) {
30 $message = $topic->consume(0, 120*10000);
31 switch ($message->err) {
32 case RD_KAFKA_RESP_ERR_NO_ERROR:
33 var_dump($message);
34 break;
35 case RD_KAFKA_RESP_ERR__PARTITION_EOF:
36 echo "No more messages; will wait for more\n";
37 break;
38 case RD_KAFKA_RESP_ERR__TIMED_OUT:
39 echo "Timed out\n";
40 break;
41 default:
42 throw new \Exception($message->errstr(), $message->err);
43 break;
44 }
45}
46?>步骤三：编译并运行
运行上述两个代码文件。
Bash
1# 启动消费者
2php consumer.php
3# 启动生产者
4php producer.php步骤四：查看集群监控
查看消息是否发送成功或消费成功有两种方式：
- 查看程序输出日志。
- 在专享版消息服务 for Kafka控制台查看集群监控，获取集群生产、消息情况。
推荐使用第二种方式，下面介绍如何查看集群监控。
（1）在专享版消息服务 for Kafka的控制台页面找到需要连接的集群，点击集群名称进入『集群详情』页面。
（2）页面跳转后，进入左侧边中的『集群详情』页面。
（3）点击左侧边栏中的『集群监控』，进入『集群监控』页面。
（4）通过查看『集群监控』页面，提供的不同纬度的监控信息（集群监控、节点监控、主题监控、消费组监控），即可获知集群的生产和消费情况。
集群监控的具体使用请参考：集群监控
