1.3 生产与消费

由 1.1 节的内容可知，生产者将消息发送至 Kafka 的主题中，或者更加确切地说应该是主题的分区中，而消费者也是通过订阅主题从而消费消息的。在演示生产与消费消息之前，需要创建一个主题作为消息的载体。

Kafka提供了许多实用的脚本工具，存放在$KAFKA_HOME的bin目录下，其中与主题有关的就是 kafka-topics.sh 脚本，下面我们用它演示创建一个分区数为 4、副本因子为 3 的主题topic-demo，示例如下：

其中--zookeeper指定了Kafka所连接的ZooKeeper服务地址，--topic指定了所要创建主题的名称，--replication-factor 指定了副本因子，--partitions 指定了分区个数，--create是创建主题的动作指令。

还可以通过--describe展示主题的更多具体信息，示例如下：

创建主题topic-demo之后我们再来检测一下Kafka集群是否可以正常地发送和消费消息。$KAFKA_HOME/bin 目录下还提供了两个脚本 kafka-console-producer.sh 和 kafka-console-consumer.sh，通过控制台收发消息。首先我们打开一个shell终端，通过kafka-console-consumer.sh脚本来订阅主题topic-demo，示例如下：

其中--bootstrap-server指定了连接的Kafka集群地址，--topic指定了消费者订阅的主题。目前主题topic-demo尚未有任何消息存入，所以此脚本还不能消费任何消息。

我们再打开一个shell终端，然后使用kafka-console-producer.sh脚本发送一条消息“Hello，Kafka！”至主题topic-demo，示例如下：

其中--broker-list指定了连接的Kafka集群地址，--topic指定了发送消息时的主题。示例中的第二行是通过人工键入的方式输入的，按下回车键后会跳到第三行，即“＞”字符处。此时原先执行 kafka-console-consumer.sh脚本的 shell终端中出现了刚刚输入的消息“Hello，Kafka！”，示例如下：

读者也可以通过输入一些其他自定义的消息来熟悉消息的收发及这两个脚本的用法。不过这两个脚本一般用来做一些测试类的工作，在实际应用中，不会只是简单地使用这两个脚本来做复杂的与业务逻辑相关的消息生产与消费的工作，具体的工作还需要通过编程的手段来实施。下面就以Kafka自身提供的Java客户端来演示消息的收发，与Kafka的Java客户端相关的Maven依赖如下：

要往Kafka中写入消息，首先要创建一个生产者客户端实例并设置一些配置参数，然后构建消息的ProducerRecord对象，其中必须包含所要发往的主题及消息的消息体，进而再通过生产者客户端实例将消息发出，最后可以通过 close（）方法来关闭生产者客户端实例并回收相应的资源。具体的示例如代码清单1-1所示，与脚本演示时一样，示例中仅发送一条内容为“Hello，Kafka！”的消息到主题topic-demo。

代码清单1-1 生产者客户端示例代码

对应的消费消息也比较简单，首先创建一个消费者客户端实例并配置相应的参数，然后订阅主题并消费即可，具体的示例代码如代码清单1-2所示。

代码清单1-2 消费者客户端示例代码

通过这些示例，相信各位读者对Kafka应该有了初步的认识。这仅仅是一个开始，要正确、灵活地运用好Kafka还需要对它进行深入探索，包括生产者和消费者客户端的使用细节及原理、服务端的使用细节及原理、运维、监控等，每一个方面都等着读者去一一攻破。