kafka学习笔记（三、消费者Consumer使用教程——消费性能多线程提升思考）

1.简介

KafkaConsumer是非线程安全的，它定义了一个acquire()方法来检测当前是否只有一个线程在操作，如不是则会抛出ConcurrentModifcationException异常。

acquire()可以看做是一个轻量级锁，它仅通过线程操作计数标记的方式来检测线程是否发生了并发操作，以此保证只有一个线程在操作。acquire()方法和release()方法成对出现，表示相应的加锁操作和解锁操作。

KafkaConsumer虽然是单线程的执行方式，但是在某些情况下如：生产者发送消息的速度远大于消费者消费的速度，这样长时间可能会造成消息的丢失，此时我们就需要消费者采用多线程消费的方式增加消费速度。

2.多线程实现的方式

2.1.线程封闭多线程

即为每个线程实例化一个KafkaConsumer，如图所示，一个线程对应一个KafkaConsumer实例，所有的消费线程都属于同一个消费者组。

这种方式的并发度受限于分区的实际个数。

实现代码示例：

public class kafkaConsumer1 {
    public void ConsuermMultithread1() {
        Properties props = initConsifg(); // 此处初始化消费者配置参数省略
        int consumerThreadNum = 5;
        for (int i = 0; i  
2.1.消息处理模块多线程
 
此方法是对上面方法的进一步优化，在消息处理模块增加多线程来处理消息，进一步提升消息消费的速度。
 
public class kafkaConsumer1 {
    public void ConsuermMultithread1() {
        Properties props = initConsifg(); // 此处初始化消费者配置参数省略
        int consumerThreadNum = 5;
        for (int i = 0; i  
 
 
此方法需要引入一个共享的offsets来参与提交。