公网SASL_SSL方式生产和消费消息
更新时间：2024-06-18
在 Kafka 集群所在 VPC 网络外访问，使用 SASL_SSL 协议接入，接入点可以在 集群详情 页面查看。
环境准备
- 安装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 | 接入点信息 |
| ssl.ca.location | ca.pem证书文件所在路径 |
| sasl.username | 用户管理中创建用户的用户名 |
| sasl.password | 用户管理中创建用户的密码 |
| topic_name | 主题名称 |
| group_id | 消费组id |
| sasl.mechanism | SCRAM-SHA-512或者PLAIN |
生产者代码示例
创建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_ssl');
8// 证书文件路径（证书文件请参考"接入点查看"文档）
9$conf->set('ssl.ca.location',__DIR__.'/ssl.cert/ca.pem');
10// SASL 机制
11$conf->set('sasl.mechanism','SCRAM-SHA-512');
12// SASL 用户名
13$conf->set('sasl.username','alice');
14// SASL 密码
15$conf->set('sasl.password','alice1234!');
16
17$producer = new RdKafka\Producer($conf);
18
19// 指定topic
20$topic = $producer->newTopic("topic_name");
21
22for ($i = 0; $i < 10; $i++) {
23 // RD_KAFKA_PARTITION_UA自动选择分区，message指定想要发送的消息
24 $topic->produce(RD_KAFKA_PARTITION_UA, 0, "Message $i : php sasl_ssl");
25 $producer->poll(0);
26}
27
28for ($flushRetries = 0; $flushRetries < 10; $flushRetries++) {
29 $result = $producer->flush(10000);
30 if (RD_KAFKA_RESP_ERR_NO_ERROR === $result) {
31 break;
32 }
33}
34
35if (RD_KAFKA_RESP_ERR_NO_ERROR !== $result) {
36 throw new \RuntimeException('Was unable to flush, messages might be lost!');
37}消费者代码示例
创建KafkaConsumerDemo.php文件，具体代码示例如下：
Shell
1<?php
2
3$conf = new RdKafka\Conf();
4// 接入协议
5$conf->set('security.protocol','sasl_ssl');
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// 证书文件路径
13$conf->set('ssl.ca.location',__DIR__.'/ssl.cert/ca.pem');
14// 消费组 id
15$conf->set('group.id', 'php-group');
16
17$rk = new RdKafka\Consumer($conf);
18// 接入点
19$rk->addBrokers("接入点");
20
21$topicConf = new RdKafka\TopicConf();
22$topicConf->set('auto.commit.interval.ms', 100);
23$topicConf->set('offset.store.method', 'broker');
24$topicConf->set('auto.offset.reset', 'earliest');
25
26//订阅topic
27$topic = $rk->newTopic("topic_name", $topicConf);
28
29$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);
30
31while (true) {
32 $message = $topic->consume(0, 120*10000);
33 switch ($message->err) {
34 case RD_KAFKA_RESP_ERR_NO_ERROR:
35 var_dump($message);
36 break;
37 case RD_KAFKA_RESP_ERR__PARTITION_EOF:
38 echo "No more messages; will wait for more\n";
39 break;
40 case RD_KAFKA_RESP_ERR__TIMED_OUT:
41 echo "Timed out\n";
42 break;
43 default:
44 throw new \Exception($message->errstr(), $message->err);
45 break;
46 }
47}步骤四：编译并运行
运行上述两个代码文件。
Bash
1# 启动消费者
2php consumer.php
3# 启动生产者
4php producer.php步骤五：查看集群监控
查看消息是否发送成功或消费成功有两种方式：
- 查看程序输出日志。
- 在专享版消息服务 for Kafka控制台查看集群监控，获取集群生产、消息情况。
推荐使用第二种方式，下面介绍如何查看集群监控。
（1）在专享版消息服务 for Kafka的控制台页面找到需要连接的集群，点击集群名称进入『集群详情』页面。
（2）页面跳转后，进入左侧边中的『集群详情』页面。
（3）点击左侧边栏中的『集群监控』，进入『集群监控』页面。
（4）通过查看『集群监控』页面，提供的不同纬度的监控信息（集群监控、节点监控、主题监控、消费组监控），即可获知集群的生产和消费情况。
集群监控的具体使用请参考：集群监控
