操作场景
该任务以 Python 客户端为例，指导您使用公网 SASL_PLAINTEXT 方式接入消息队列 CKafka 版并收发消息。
前提条件
操作步骤
步骤1：准备工作
1. 创建接入点。
1.1 在 实例列表 页面，单击目标实例 ID，进入实例详情页。
1.2 在 基本信息 > 接入方式 中，单击添加路由策略，在打开窗口中选择：
路由类型：公网域名接入, 接入方式：SASL_PLAINTEXT。
2. 创建角色。
在 ACL 策略管理下的用户管理页面新建角色，设置密码。
3. 创建 Topic。
在控制台 topic 管理页面新建 Topic（参见 创建 Topic）。
4. 添加 Python 依赖库。
执行以下命令安装：
pip install kafka-python
步骤2：生产消息
1. 修改生产消息程序
producer.py 中配置参数。producer = KafkaProducer(bootstrap_servers = ['xx.xx.xx.xx:port'],api_version = (1, 1),## SASL_PLAINTEXT 公网接入#security_protocol = "SASL_PLAINTEXT",sasl_mechanism = "PLAIN",sasl_plain_username = "instanceId#username",sasl_plain_password = "password",)message = "Hello World! Hello Ckafka!"msg = json.dumps(message, ensure_ascii=False).encode()producer.send('topic_name', value = msg)print("produce message " + message + " success.")producer.close()
参数 | 描述 |
bootstrap_servers | 接入网络，在控制台的实例详情页面接入方式模块的网络列复制。  |
sasl_plain_username | |
sasl_plain_password | 用户密码，在 CKafka 控制台实例详情页面ACL策略管理下的用户管理创建用户时设置。 |
topic_name | Topic 名称，您可以在控制台上topic管理页面复制。  |
2. 编译并运行 producer.py。
3. 查看运行结果。
4. 在 CKafka 控制台 的 topic管理页面，选择对应的 Topic ， 单击更多 > 消息查询，查看刚刚发送的消息。
步骤3：消费消息
1. 修改消费消息程序 consumer.py 中配置参数。
consumer = KafkaConsumer('topic_name',group_id = "group_id",bootstrap_servers = ['xx.xx.xx.xx:port'],api_version = (1,1),## SASL_PLAINTEXT 公网接入#security_protocol = "SASL_PLAINTEXT",sasl_mechanism = 'PLAIN',sasl_plain_username = "instanceId#username",sasl_plain_password = "password",)for message in consumer:print ("Topic:[%s] Partition:[%d] Offset:[%d] Value:[%s]" %(message.topic, message.partition, message.offset, message.value))
参数 | 描述 |
bootstrap_servers | 接入网络，在控制台的实例详情页面接入方式模块的网络列复制。  |
group_id | 消费者的组 ID，根据业务需求自定义。 |
sasl_plain_username | 用户名，格式为 实例 ID + # + 用户名。实例 ID 在CKafka 控制台的实例详情页面的基本信息获取，用户在ACL策略管理
下的用户管理创建用户时设置。 |
sasl_plain_password | 用户名密码，在 CKafka 控制台实例详情页面ACL策略管理下的用户管理创建用户时设置。 |
topic_name | Topic 名称，您可以在控制台上 topic管理页面复制。  |
2. 编译并运行 consumer.py。
3. 查看运行结果。
4. 在 CKafka 控制台 的 Consumer Group 页面，选择对应的消费组名称，在主题名称输入 Topic 名称，单击查询详情，查看消费详情。
