Consumer 多线程
Consumer 多线程有2中形式：
- 每个线程维护一个KafkaConsumer
- 单KafkaConsumer实例+多worker线程
本文实践第一种方式，创建多个线程消费 topic，每个线程都会创建属于该线程的 consumer。
下面示例中有3个class：
-
ConsumerRunnable
消费线程类，执行真正的消费任务。
-
ConsumerGroup
消费线程管理类，创建多个线程类执行消费任务。
-
ConsumerMain
测试主方法类。
示例
- 创建项目
$ mvn archetype:generate -DgroupId=com.kafkademo -DartifactId=KafkaConsumerMultiThread01 -DarchetypeArtifactId=maven-archetype-quickstart -DinteractiveMode=false
- 添加依赖
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.0.0</version>
</dependency>
...
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.8</source>
<target>1.8</target>
</configuration>
</plugin>
</plugins>
</build>
- 代码
ConsumerRunnable
package com.kafkademo;
import java.util.Arrays;
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 ConsumerRunnable implements Runnable {
// 每个线程维护私有的KafkaConsumer 实例
private final KafkaConsumer<String, String> consumer;
public ConsumerRunnable(String brokerList, String groupId, String topic) {
Properties props = new Properties();
props.put("bootstrap.servers", brokerList);
props.put("group.id", groupId);
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("session.timeout.ms", "30000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
this.consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList(topic));
}
@Override
public void run() {
while (true) {
ConsumerRecords<String, String> records = consumer.poll(200);
for (ConsumerRecord<String, String> record : records) {
System.out.println("[Thread] " + Thread.currentThread().getName() + ", [partition] "
+ record.partition() + ", [offset] " + record.offset() + ", [value] " + record.value());
}
}
}
}
ConsumerGroup
package com.kafkademo;
import java.util.ArrayList;
import java.util.List;
public class ConsumerGroup {
private List<ConsumerRunnable> consumers;
public ConsumerGroup(int consumerNum, String groupId, String topic, String brokerList) {
consumers = new ArrayList<>(consumerNum);
for (int i = 0; i < consumerNum; ++i) {
ConsumerRunnable consumerThread = new ConsumerRunnable(brokerList, groupId, topic);
consumers.add(consumerThread);
}
}
public void execute() {
for(ConsumerRunnable task : consumers) {
new Thread(task).start();
}
}
}
ConsumerTest
package com.kafkademo;
public class ConsumerTest {
public static void main(String[] args) {
String brokerList = "localhost:9092";
String groupId = "testGroup1";
String topic = "test-topic1";
int consumerNum = 3;
ConsumerGroup consumerGroup = new ConsumerGroup(consumerNum, groupId, topic, brokerList);
consumerGroup.execute();
}
}
- 运行
先创建 topic：
$ bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic test-topic1
运行：
$ mvn compile
$ mvn exec:java -Dexec.mainClass="com.kafkademo.ConsumerTest"
运行后会进入等待状态，等待 test-topic1 中有新的消息。
- 测试
生产消息：
$ bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test-topic1
>1
>2
>3
>4
ConsumerTest 会输出消费信息：
[Thread] Thread-1, [partition] 0, [offset] 0, [value] 1
[Thread] Thread-3, [partition] 2, [offset] 0, [value] 2
[Thread] Thread-2, [partition] 1, [offset] 0, [value] 3
[Thread] Thread-1, [partition] 0, [offset] 1, [value] 4
