一、Kafka概述
Apache Kafka是一个分布式发布 - 订阅消息系统和一个强大的队列，可以处理大量的数据，并使您能够将消息从一个端点传递到另一个端点。 Kafka适合离线和在线消息消费。 Kafka消息保留在磁盘上，并在群集内复制以防止数据丢失。 Kafka构建在ZooKeeper同步服务之上。 Kafka支持低延迟消息传递，并在出现机器故障时提供对容错的保证。 它具有处理大量不同消费者的能力。 Kafka非常快，执行2百万写/秒。 Kafka将所有数据保存到磁盘，这实质上意味着所有写入都会进入操作系统(RAM)的页面缓存。 这使得将数据从页面缓存传输到网络套接字非常有效。
二、Kafka示例环境搭建
本示例搭建的Kafka环境为示例代码演示环境，生产环境一般为集群配置。
官网下载kafka并解压：Apache Kafka
启动zookeeper：Kafka依赖zookeeper服务，cd到Kafka解压目录，执行如下命令启动zookeeper。
bin/zookeeper-server-start.sh config/zookeeper.properties
启动Kafka：配置config/server.properties，增加listeners配置如下，其中IP地址为本机IP，保存配置。
listeners=PLAINTEXT://192.168.2.14:9092
执行bin/kafka-server-start.sh config/server.properties启动Kafka。
安装Kafka可视化工具：Offset Explorer (kafkatool.com)。
如果正常工作环境为Windows系统，可以下载安装Kafka可视化工具，方便查看Kafka Topic消息及消费组信息等。
安装完毕后，打开Offset Explorer，参照如下配置新建连接，可以查看Brokers、Topic以及Consumers信息，初次打开的话内容应该为空，没有新建的Topic、也没有Consumer。
三、多个消费组独立消费同一个消息队列示例
Kafka一个很重要的特性就是，只需写入一次消息，可以支持任意多的应用读取这个消息。 换句话说，每个应用都可以读到全量的消息。为了使得每个应用都能读到全量消息，应用需要有不同的消费组。接下来通过java编写代码来演示该能力。
创建java maven工程，pom.xml添加Kafka依赖。
<dependencies> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>2.8.1</version> </dependency> </dependencies>
创建ProducerDemo类，代码及注释如下：
package com.demo.kafka;
import java.util.Properties;
import java.util.Scanner;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
public class ProducerDemo {
// Kafka消息主题
public final static String TOPIC = "DEMO";
// Kafka Broker Server
public final static String BROKER_LIST = "192.168.2.14:9092";
public static void main(String[] args) {
Properties props = new Properties();
// 设置key序列化器
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// 设置value序列化器
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// 设置重试次数
props.put(ProducerConfig.RETRIES_CONFIG, 10);
// 设置集群服务地址
props.put("bootstrap.servers", BROKER_LIST);
// 根据上述属性配置创建生产者
KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
while (true) {
// 控制台接收用户输入消息，回车发送
Scanner input = new Scanner(System.in);
if (input.hasNextLine()) {
String message = input.nextLine();
ProducerRecord<String, String> record = new ProducerRecord<String, String>(TOPIC, message);
try {
producer.send(record);
} catch (Exception e) {
e.printStackTrace();
producer.close();
}
}
}
}
}创建ConsumerDemo类，代码及注释如下：
package com.demo.kafka;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
public class ConsumerDemo extends Thread {
// Kafka消息主题
public final static String TOPIC = "DEMO";
// Kafka Broker Server
public final static String BROKER_LIST = "192.168.2.14:9092";
public static void main(String[] args) {
Properties props = new Properties();
// 设置key反序列化器
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
// 设置value反序列化器
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
// 设置集群服务地址
props.put("bootstrap.servers", BROKER_LIST);
// 创建消费组A
props.put("group.id", "group.A");
KafkaConsumer<String, String> consumerA = new KafkaConsumer<>(props);
// 创建消费组B
props.put("group.id", "group.B");
KafkaConsumer<String, String> consumerB = new KafkaConsumer<>(props);
// 创建消费组C
props.put("group.id", "group.C");
KafkaConsumer<String, String> consumerC = new KafkaConsumer<>(props);
// 消费组A、B、C订阅同一个消息队列
consumerA.subscribe(Collections.singletonList(TOPIC));
consumerB.subscribe(Collections.singletonList(TOPIC));
consumerC.subscribe(Collections.singletonList(TOPIC));
while (true) {
// 消费组A消费TOPIC中的消息，拉取间隔为1s
ConsumerRecords<String, String> records = consumerA.poll(Duration.ofMillis(1000));
for (ConsumerRecord<String, String> record : records) {
System.out.println("consumerA消费消息：");
System.out.printf("offset = %d, value = %s", record.offset(), record.value());
System.out.println();
}
// 消费组B消费TOPIC中的消息，拉取间隔为1s
records = consumerB.poll(Duration.ofMillis(1000));
for (ConsumerRecord<String, String> record : records) {
System.out.println("consumerB消费消息：");
System.out.printf("offset = %d, value = %s", record.offset(), record.value());
System.out.println();
}
// 消费组C消费TOPIC中的消息，拉取间隔为1s
records = consumerC.poll(Duration.ofMillis(1000));
for (ConsumerRecord<String, String> record : records) {
System.out.println("consumerC消费消息：");
System.out.printf("offset = %d, value = %s", record.offset(), record.value());
System.out.println();
}
}
}
}运行ConsumerDemo.main()，该类共创建了三个消费组consumerA、consumerB、consumerC，均订阅了同一个消息主题DEMO，并持续在控制台打印消费的消息。
- 运行ProducerDemo.main()，该类创建了一个生产者并持续接收控制台用户的输入作为消息内容，并向消息主题DEMO发送消息。
现在，试着在控制台输入一条消息并回车。
然后查看ConsumerDemo页签查看各消费组消费的消息信息，可以发现三个消费组都消费了同一个主题信息。
- 通过Offset Explorer可视化工具也可以查看到对应TOPIC下的消息，以及创建的消费组。
