《Apache Kafka 实战》笔记 - 5.8.1 Consumer 多线程01

Consumer 多线程

Consumer 多线程有2中形式：

1. 每个线程维护一个KafkaConsumer
2. 单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