本文介绍通过 SASL_PLAINTEXT 接入点连接 Kafka 实例，进行消息生产和消息消费的操作步骤。

背景信息

消息队列 Kafka版提供 SASL/PLAIN 协议的安全访问方式，即 SASL_PLAINTEXT 接入点。通过 SASL_PLAINTEXT 接入点连接实例时，需要通过 SASL 用户名及密码鉴权后才能访问 Kafka 实例。消息队列 Kafka版提供 PLAIN 机制和 SCRAM 机制供您访问和接入，创建实例时可以同步创建 SASL 用户，您也可以在创建实例后按需创建 SASL 用户，在访问实例时按需通过不同机制生产或消费消息。
默认情况下可以通过 VPC 内访问 Kafka 实例，开启公网访问功能之后，还可以通过公网访问实例。两种网络类型均可以通过 SASL_PLAINTEXT 接入点进行接入。如果公网环境下访问实例需要通过 SSL 认证，推荐选择 SASL_SSL 接入点，详细接入方式请参考使用 SASL_SSL 接入点连接实例。

前提条件

• 已获取 SASL_PLAINTEXT 接入点信息，包括连接地址和端口号。详细信息请参考查看接入点。
• 已创建 Topic。操作步骤请参考创建Topic。
• 如果通过公网访问实例，建议实例绑定的 EIP 带宽上限大于预估的公网业务流量峰值。详细信息请参考配置公网访问。
• 环境中已成功安装 JDK、配置环境变量，并下载 Kafka 开源客户端，例如 Kafka 2.2.2 客户端。

配置鉴权

在 Kafka 客户端包的./config目录中，修改consumer.properties和producer.properties文件。
请根据 SASL 机制类型，分别在文件中增加以下内容。详细的配置文件说明请参考配置文件。

• 使用 PLAIN 机制时：
  在 consumer.properties 和 producer.properties 文件中增加以下内容。其中 username 和 password 须填写 PLAIN 用户的用户名和密码。

  sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="{$plainusername}" password="{$plainpwxxx}";        
  sasl.mechanism=PLAIN
  security.protocol=SASL_PLAINTEXT
  

• 使用 SCRAM-SHA-256 机制时：
  在 consumer.properties 和 producer.properties 文件中增加以下内容。其中 username 和 password 须填写 SCRAM 用户的用户名和密码。

  sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="{$scramusername}" password="{$scrampw}";        
  sasl.mechanism=SCRAM-SHA-256
  security.protocol=SASL_PLAINTEXT
  

生产消息

1. 解压 Kafka 客户端文件。

2. 在 ./bin 目录下，打开终端。

3. 执行以下命令启动生产者，开始生产消息。

   bash kafka-console-producer.sh --broker-list ${SASL接入点} --topic ${Topic名称}  --producer.config ../config/producer.properties
   

   参数	说明
   SASL接入点	控制台实例详情页面获取的 SASL_PLAINTEXT 接入点的域名和端口号。
   Topic名称	Kafka 实例的 Topic 名称。

4. 输入需要生产的消息内容，按回车键发送此消息到 Kafka 实例。
   光标下输入的每一行内容都将封装到消息中发送到 Kafka 实例。如需停止生产，可以使用 Ctrl+C 命令退出消息生产。

   [root@kafkaecs bin]# bash kafka-console-producer.sh --broker-list kafka-cnngc7an0qp****.kafka.ivolces.com:9093 --topic mytopic --producer.config ../config/producer.properties
   >Hello world!
   >This is a Kafka message!
   >^C[root@kafkaecs bin]# 
   

消费消息

1. 打开另一个终端，执行如下命令消费消息。
   如需停止消费，可以使用 Ctrl+C 命令退出消息消费。

   bash kafka-console-consumer.sh --bootstrap-server ${SASL接入点}  --topic ${Topic名称} --group ${消费组名称} --from-beginning --consumer.config ../config/consumer.properties
   

   参数	说明
   SASL接入点	控制台实例详情页面获取的 SASL_PLAINTEXT 接入点的域名和端口号。
   Topic名称	Kafka 实例的 Topic 名称。
   消费组名称	自定义的消费组名称。 如果您已在配置文件中指定了消费组名称，请确认命令行中配置了相同的名称，否则可能导致消费失败。

2. 成功操作后，消费者可以收到生产者发送的消息，内容与生产端的消息相同。

   [root@kafka bin]# bash kafka-console-consumer.sh --bootstrap-server kafka-cnngc7an0qpv****.kafka.ivolces.com:9093 --topic mytopic --group grouptest --from-beginning --consumer.config ../config/consumer.properties
   Hello world!
   This is a Kafka message!
   ^CProcessed a total of 2 messages
   [root@kafka bin]#