前提条件

• 已配置正确的安全组。
• 已获取连接Kafka实例的地址。

说明
使用内网通过同一个VPC访问，实例端口为8090，实例连接地址从控制台集群信息菜单处获取。

• 如果Kafka实例未开启自动创建Topic功能，在连接实例前，请先创建Topic。
• 已下载Kafka命令行工具2.8.2版本，确保Kafka实例版本与命令行工具版本相同。
• 已创建弹性云服务器，如果使用内网通过同一个VPC访问实例，请设置弹性云服务器的VPC、子网、安全组与Kafka实例的VPC、子网、安全组一致。

命令行模式连接实例

以下操作命令以Linux系统为例进行说明：

1.解压Kafka命令行工具。

进入文件压缩包所在目录，然后执行tar -zxf [kafka_tar]命令解压文件。

其中，[kafka_tar]表示命令行工具的压缩包名称。
例如：tar -zxf kafka_2.13-2.8.2.tgz
2.进入Kafka命令行工具的“/bin”目录下。

注意
Windows系统下需要进入“/bin/windows”目录下。

3.执行如下命令进行生产消息。

./kafka-console-producer.sh --broker-list ${连接地址} --topic ${Topic名称}

说明
连接地址：从前提条件中获取的连接地址。
Topic名称：Kafka实例下创建的Topic名称。如果Kafka实例开启了自动创建Topic功能，此参数值可以填写已创建的Topic名称，也可以填写未创建的Topic名称。

本文获取的Kafka实例内网连接地址为“10.10.3.214:8090,10.10.3.126:8090”。执行完命令后输入内容，按“Enter”发送消息到Kafka实例，输入的每一行内容都将作为一条消息发送到Kafka实例。

[root@ctyun-kafka bin]# ./kafka-console-producer.sh --broker-list 10.10.3.214:8090,10.10.3.126:8090  --topic topic-demo
Hello
DMS
Kafka!
[root@ctyun-kafka bin]# 

如需停止生产使用Ctrl+C命令退出。

4.执行如下命令消费消息。

./kafka-console-consumer.sh --bootstrap-server ${连接地址} --topic ${Topic名称} --group ${消费组名称} --from-beginning

说明
连接地址：从前提条件中获取的连接地址。
Topic名称：Kafka实例下创建的Topic名称。
消费组名称：根据您的业务需求，设定消费组名称。如果已经在配置文件中指定了消费组名称，请确保命令行中的消费组名称与配置文件中的相同，否则可能消费失败。消费组名称开头包含特殊字符，例如#号“#”时，监控数据无法展示。

示例如下：

[root@ctyun-kafka bin]#  ./kafka-console-consumer.sh --bootstrap-server 10.10.3.214:8090,10.10.3.126:8090 --topic topic-demo --group order-test --from-beginning
Kafka!
DMS
Hello
Processed a total of 3 messages
[root@ctyun-kafka bin]# 

如需停止消费使用Ctrl+C命令退出。