一、准备工作

进入实战之前先熟悉一下topic的相关命令，使用终端命令查询创建一个新topic，用于后期实战；
特别注意：以下命令全部依据kafka文件目录中操作；
如果尚未安装kafka，请移步《centos7系统安装kafka》

• 查看操作主题命令参数
  命令：./bin/kafka-topics.sh
  日常工作中常用的命令如下表：

• 查看当前服务器中的所有topic
  命令：bin/kafka-topics.sh --list --bootstrap-server localhost:9092
  

• kafka生产者数据阻塞、创建topic
  命令：bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --partitions 1 --replication-factor 1 --topic kafka-test

选项说明：

1. –topic 定义 topic 名
2. –replication-factor 定义副本数
3. –partitions 定义分区数

• 查看topic详情
  命令：bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic kafka-test --describe
  

• 修改topic分区
  命令：bin/kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic kafka-test --partitions 3

注意：kafka一旦增加分区号，不支持减少分区数；

二、终端命令

生产者命令

• 查看操作者命令参数
  命令：bin/kafka-console-producer.sh

• 生产者发送消息
  命令：bin/kafka-console-producer.sh --broker-list localhost:9092 --topic kafka-test

消费者命令

• 查看操作消费者命令参数
  命令：bin/kafka-console-consumer.sh

kafka事务消息，消费者消费消息

• 消费主题中的消息

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic kafka-test

• 主题中所有的数据都读取出来包括历史数据。

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic kafka-test

执行结果如图：

三、Java实践

正式进入生产者代码实践之前，首先列举出生产者方大致的参数列表如下：

搭建项目

• 1、引入kafka依赖

<dependencies><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>3.0.0</version></dependency>
</dependencies>

• 2、打开消费者监听topic命令

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic kafka-test

异步发送-无回调

代码如下：

public static void main(String[] args) {// 1. 创建kafka生产者配置对象Properties properties = new Properties();// 2. 给 kafka 配置对象添加配置信息：bootstrap.serversproperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "XXXXXXX:9092");// key,value 序列化（必须）：key.serializer，value.serializerproperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");// 3. 创建 kafka 生产者对象KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);// 4. 调用 send 方法,异步发送消息，无回调接口for (int i = 0; i < 5; i++) {kafkaProducer.send(new ProducerRecord<>("kafka-test", "producerAsyncMessage-" + i));}// 5. 关闭资源kafkaProducer.close();}

执行结果如图：

异步发送-有回调

public static void main(String[] args) throws InterruptedException {// 1. 创建Kafka生产者的配置对象Properties properties = new Properties();// 2. 给kafka配置对象添加配置信息properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "xxxxxx:9092");// 3. key 序列化 key.serializer，value.serializerproperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());// 4. value 序列化 value.serializerproperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());// 5. 创建kafka生产者对象KafkaProducer<String,String> kafkaProducer = new KafkaProducer<String, String>(properties);for (int i = 0; i < 5; i++) {kafkaProducer.send(new ProducerRecord<>("kafka-test", "asyncCallbackMessage-" + i), new Callback() {// 该方法在Producer 收到 ack 时调用，为异步调用@Overridepublic void onCompletion(RecordMetadata recordMetadata, Exception e) {if (e == null) {// 没有异常，输出信息到控制台System.out.println("topic：" + recordMetadata.topic() + " -> " + " 分区 " + recordMetadata.partition());}else {e.printStackTrace();}}});// 延迟发送是否会发送到不同分区Thread.sleep(2);}// 5. 关闭资源kafkaProducer.close();}

• kafka推送数据，消费者接收消息结果
  

• 生产者接收回调结果
  

同步发送

public static void main(String[] args) throws ExecutionException, InterruptedException {// 1. 创建kafka生产者配置对象Properties properties = new Properties();// 2. 给 kafka 配置对象添加配置信息：bootstrap.serversproperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "xxxxxx:9092");// key,value 序列化（必须）：key.serializer，value.serializerproperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");// 3. 创建 kafka 生产者对象KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);// 4. 调用 send 方法,发送消息for (int i = 0; i < 5; i++) {RecordMetadata recordMetadata = kafkaProducer.send(new ProducerRecord<>("kafka-test", "syncMessage-" + i)).get();System.out.println("send response result,topic：" + recordMetadata.topic() + ", " + " 分区:" + recordMetadata.partition());}// 5. 关闭资源kafkaProducer.close();}

• 消费者接收消息结果

[root@VM-4-8-centos kafka]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic kafka-test
producerAsyncMessage-0
producerAsyncMessage-1
producerAsyncMessage-2
producerAsyncMessage-3
producerAsyncMessage-4
asyncCallbackMessage-0
asyncCallbackMessage-1
asyncCallbackMessage-2
asyncCallbackMessage-3
asyncCallbackMessage-4
syncMessage-0
syncMessage-1
syncMessage-2
syncMessage-3
syncMessage-4

• 生产者接收发送结果