# 使用Kafka生产消费数据
#### 操作场景
用户可以通过MRS集群Kafka客户端在Kafka Topic中产生和消费消息。
#### 前提条件
- 已安装客户端，例如安装目录为"/opt/client"，以下操作的客户端目录只是举例，请根据实际安装目录修改。 下载并安装集群客户端的具体操作，请参考[安装MRS集群客户端](https://support.huaweicloud.com/usermanual-mrs/mrs_01_0089.html)。
  

- 启用Kerberos认证的集群，需要提前在Manager中创建业务用户，用户拥有在Kafka主题中执行相应操作的权限。 创建用户相关操作请参考[创建Kafka用户并绑定角色](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_1032.html#mrs_01_1032__section15172119027)。
  
- 已存在待操作的Topic，相关操作可参考[创建Kafka Topic](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_24136.html)。
 
#### 使用Kafka客户端生产消费数据
1. 登录MRS集群Manager。 
   登录集群Manager具体操作，请参考[访问MRS集群Manager](https://support.huaweicloud.com/usermanual-mrs/mrs_01_0128.html)。
   
   
2. 获取Kafka节点业务IP及端口。 
   1. 选择"集群 \> 服务 \> Kafka \> 实例"，查看并记录任意一个Broker角色实例的业务IP地址。
   
   2. 选择"配置 \> 全部配置"，并根据集群模式获取端口号：
      - 集群已启用Kerberos认证（安全模式）：搜索"sasl.port"参数，查看并记录端口号，默认为21007。
      
      - 集群未启用Kerberos认证（普通模式）：搜索"port"参数，查看并记录端口号，默认为9092。
       
   
   
   
   
3. 以客户端安装用户，登录安装客户端的节点。
4. 执行以下命令，切换到客户端安装目录，例如安装目录为"/opt/client"，具体以实际替换。 
   ```
   cd /opt/client
   ```
   
   
5. 执行以下命令配置环境变量。 
   ```
   source bigdata_env
   ```
   
   
6. 执行以下命令，进行用户认证。（集群未启用Kerberos认证（普通模式）时跳过此步骤） 
   ```
   kinit 组件业务用户
   ```
   
   
7. 执行以下命令进入Kafka客户端"bin"目录。 
   ```
   cd Kafka/kafka/bin
   ```
   
   
8. 执行以下命令使用客户端工具查看帮助并使用。 
   - Kafka消息发布工具：
     ```
     ./kafka-console-producer.sh
     ```
     
   
   - Kafka消息读取工具：
     ```
     ./kafka-console-consumer.sh
     ```
     
   
   
   
   
9. 根据业务需要，管理Kafka主题中的消息。 
   - 在主题中产生消息
     ```
     sh kafka-console-producer.sh --broker-list Broker角色实例所在节点的IP地址:21007 --topic Topic名称 --producer.config /opt/client/Kafka/kafka/config/producer.properties
     ```
     Topic需提前创建，用户可以输入指定的内容作为生产者产生的消息，输入完成后按回车发送消息。如果需要结束产生消息，使用"Ctrl + C"退出任务。
     
   
   - 消费主题中的消息 重新打开一个客户端连接，执行以下命令消费主题中的消息。
     ```
     cd /opt/client/Kafka/kafka/bin
     ```
     ```
     source /opt/client/bigdata_env
     ```
     ```
     sh kafka-console-consumer.sh --topic Topic名称 --bootstrap-server Broker角色实例所在节点的IP地址:21007  --consumer.config /opt/client/Kafka/kafka/config/consumer.properties
     ```
     - 配置文件中"group.id"指定的消费者组默认为"example-group1"。用户可根据业务需要，自定义其他消费者组。每次消费时生效。
     
     - 执行命令时默认会读取当前消费者组中未被处理的消息。如果在配置文件指定了新的消费者组且命令中增加参数"--from-beginning"，则会读取所有Kafka中未被自动删除的消息。
      
   
   
   
   
 
#### 相关文档
- 关于Kafka UI页面详细介绍及操作，请参考[登录KafkaUI界面](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_24135.html)。
- 如果需要查看生产消费详情，请参考[查看Kafka数据生产消费详情](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_24138.html)。