kafka系列之Consumer 消费模式(10)

悠悠 2022-11-07 14:39 382阅读 0赞

## 消费模式 ##

### Standalone Consumer(独立消费者) ###

*  和消费者组相同的是，它们也要配置 group.id 参数值，但和消费者组调用 KafkaConsumer.subscribe() 不同的是，独立消费者调用 KafkaConsumer.assign() 方法直接消费指定分区。
 *  使用assign的consumer就是standalone consumer，一旦给消费者分配分区之后，就可以和consumer group 一样，在循环中调用‘poll’ 方法获取数据
 *  独立消费者就是没有使用消费者组机制的消费者程序，依然还是采用group 的机制来提交offset ,但是消费者失败之后没有重平衡机制
 *  也就是说虽然大家都是在同一个组下面，但是消费者之间互不影响，这也是”独立一词大来源“

#### 使用场景 ####

1.  多数流处理框架的Kafka connector都没有使用consumer group，而是直接使用standalone consumer，因为group机制不好把控
2.  standalone consumer没有rebalance，也没有group提供的负载均衡，你需要自己实现(但是又分区发现机制)

### 多线程Consumer ###

*  Kafka Java Consumer 是单线程的设计
 *  我们说 KafkaConsumer 是单线程的设计，严格来说这是不准确的。因为，从 Kafka 0.10.1.0 版本开始，KafkaConsumer 就变为了双线程的设计，即用户主线程和心跳线程。
 *  所谓用户主线程，就是你启动 Consumer 应用程序 main 方法的那个线程，而新引入的心跳线程（Heartbeat Thread）只负责定期给对应的 Broker 机器发送心跳请求，以标识消费者应用的存活性（liveness）
 *  引入这个心跳线程还有一个目的，那就是期望它能将心跳频率与主线程调用 KafkaConsumer.poll 方法的频率分开，从而**解耦真实的消息处理逻辑与消费者组成员存活性管理**,否则会因为数据处理的问题而影响成员的存活管理。
 *  虽然有心跳线程，但实际的消息获取逻辑依然是在用户主线程中完成的。因此，在消费消息的这个层面上，我们依然可以安全地认为 KafkaConsumer 是单线程的设计
 *  老版本 Consumer(scala 的consumer) 是多线程的架构，每个 Consumer 实例在内部为所有订阅的主题分区创建对应的消息获取线程，也称 Fetcher 线程

#### ConcurrentModificationException ####

我们说过了kafka 的Consumer 是单线程设计的，也就是说它是线程不安全的，如果你在多线程中使用就会抛出`ConcurrentModificationException` 异常

下面我们从源码角度来看一下为什么会抛出这个异常，开始之前我们先演示一下报错,代码如下：

public class ModificationExceptionDemo { 
        private static KafkaConsumer consumer;
        /** * 初始化配置 */
        private static Properties initConfig() { 
            Properties props = new Properties();
            props.put("bootstrap.servers", "localhost:9092");
            props.put("group.id", "test-group");
            props.put("enable.auto.commit", false);
            props.put("session.timeout.ms", 30000);
            props.put("max.poll.records", 1000);
            props.put("max.poll.interval.ms", 5000);
            props.put("key.deserializer", StringDeserializer.class.getName());
            props.put("value.deserializer", StringDeserializer.class.getName());
            return props;
        }
    
        public static void main(String[] args) { 
            Properties pros = initConfig();
            consumer = new KafkaConsumer<String, String>(pros);
            consumer.subscribe(Arrays.asList("flink_json_source_4"));
            
            new Thread(new Runnable() { 
                @Override
                public void run() { 
                    consumer.poll(100);
                }
            }).start();
    
            new Thread(new Runnable() { 
                @Override
                public void run() { 
                    consumer.poll(100);
                }
            }).start();
    
            try { 
                TimeUnit.SECONDS.sleep(100);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    }

我们看到这个报错信息已经出来了

Exception in thread "Thread-2" java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access
    	at org.apache.kafka.clients.consumer.KafkaConsumer.acquire(KafkaConsumer.java:2244)
    	at org.apache.kafka.clients.consumer.KafkaConsumer.acquireAndEnsureOpen(KafkaConsumer.java:2228)
    	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1180)
    	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1135)
    	at com.kingcall.clients.consumer.thread.ModificationExceptionDemo$2.run(ModificationExceptionDemo.java:44)
    	at java.lang.Thread.run(Thread.java:748)

其实这里我们调用了poll方法，我么就跟着这个方法进来，看到这个方法里面调用了`acquireAndEnsureOpen` 这个方法

![image-20210312170555068][]

private void acquireAndEnsureOpen() { 
        acquire();
        if (this.closed) { 
            release();
            throw new IllegalStateException("This consumer has already been closed.");
        }
    }
    
    private void acquire() { 
        long threadId = Thread.currentThread().getId();
        if (threadId != currentThread.get() && !currentThread.compareAndSet(NO_CURRENT_THREAD, threadId))
            throw new ConcurrentModificationException("KafkaConsumer is not safe for multi-threaded access");
        refcount.incrementAndGet();
    }

其实我们看到`acquire` 就是我们的逻辑判断，判断是不是同一个线程，如果不是的则抛出异常，其实consumer 端的代码，几乎全部都调用了`acquireAndEnsureOpen` 方法

#### 多线程设计 ####

*  Consumer 获取到消息后，处理消息的逻辑是否采用多线程，完全由你决定。单线程的设计能够简化 Consumer 端的设计。Consumer 获取到消息后，处理消息的逻辑是否采用多线程，完全由你决定。
 *  KafkaConsumer 类不是线程安全的 (thread-safe)所有的网络 I/O 处理都是发生在用户主线程中，因此，你在使用过程中必须要确保线程安全。
 *  简单来说，就是你不能在多个线程中共享同一个 KafkaConsumer 实例，否则程序会抛出 ConcurrentModificationException 异常。

![image-20210306221623211][]

##### 方案一 #####

**优势**

*  实现起来简单，因为它比较符合目前我们使用 Consumer API 的习惯。我们在写代码的时候，使用多个线程并在每个线程中创建专属的 KafkaConsumer 实例就可以了。
 *  多个线程之间彼此没有任何交互，省去了很多保障线程安全方面的开销。
 *  由于每个线程使用专属的 KafkaConsumer 实例来执行消息获取和消息处理逻辑，因此，Kafka 主题中的每个分区都能保证只被一个线程处理，这样就很容易实现分区内的消息消费顺序。这对在乎事件先后顺序的应用场景来说，是非常重要的优势。

**不足**

*  每个线程都维护自己的 KafkaConsumer 实例，必然会占用更多的系统资源，比如内存、TCP 连接等。在资源紧张的系统环境中，方案 1 的这个劣势会表现得更加明显。
 *  这个方案能使用的线程数受限于 Consumer 订阅主题的总分区数。我们知道，在一个消费者组中，每个订阅分区都只能被组内的一个消费者实例所消费。假设一个消费者组订阅了 100 个分区，那么方案 1 最多只能扩展到 100 个线程，多余的线程无法分配到任何分区，只会白白消耗系统资源。
 *  当然了，这种扩展性方面的局限可以被多机架构所缓解。除了在一台机器上启用 100 个线程消费数据，我们也可以选择在 100 台机器上分别创建 1 个线程，效果是一样的。因此，如果你的机器资源很丰富，这个劣势就不足为虑了。
 *  每个线程完整地执行消息获取和消息处理逻辑。一旦消息处理逻辑很重，造成消息处理速度慢，就很容易出现不必要的 Rebalance，从而引发整个消费者组的消费停滞。这个劣势你一定要注意。我们之前讨论过如何避免 Rebalance。

##### 方案二 #####

*  与方案 1 的粗粒度不同，方案 2 将任务切分成了消息获取和消息处理两个部分，分别由不同的线程处理它们。比起方案 1，方案 2 的最大优势就在于它的高伸缩性，就是说我们可以独立地调节消息获取的线程数，以及消息处理的线程数，而不必考虑两者之间是否相互影响。如果你的消费获取速度慢，那么增加消费获取的线程数即可；如果是消息的处理速度慢，那么增加 Worker 线程池线程数即可。

**优势**

*  它的实现难度要比方案 1 大得多，毕竟它有两组线程，你需要分别管理它们。因为该方案将消息获取和消息处理分开了，也就是说获取某条消息的线程不是处理该消息的线程，因此无法保证分区内的消费顺序。举个例子，比如在某个分区中，消息 1 在消息 2 之前被保存，那么 Consumer 获取消息的顺序必然是消息 1 在前，消息 2 在后，但是，后面的 Worker 线程却有可能先处理消息 2，再处理消息 1，这就破坏了消息在分区中的顺序。还是那句话，如果你在意 Kafka 中消息的先后顺序，方案 2 的这个劣势是致命的。
 *  方案 2 引入了多组线程，使得整个消息消费链路被拉长，最终导致正确位移提交会变得异常困难，结果就是可能会出现消息的重复消费。如果你在意这一点，那么我不推荐你使用方案 2。

#### 消费者程序启动多个线程 ####

*  费者程序启动多个线程，每个线程维护专属的 KafkaConsumer 实例，负责完整的消息获取、消息处理流程。

![img][]

##### 代码实现 #####

下面我们的演示topic`flink_json_source_4`是两个分区

public class MultiConsumerMode1 { 
    
        // 在实际应用中，你可以创建多个 KafkaConsumerRunner 实例，并依次执行启动它们，以实现多线程架构,需要注意的是启的太多没有意义
        @Test
        public void test() { 
            int partitionCount = 2;
            ExecutorService executor = Executors.newFixedThreadPool(3);
            for (int i = 0; i < partitionCount; i++) { 
                executor.submit(new KafkaConsumerRunner());
            }
            try { 
                executor.awaitTermination(10, TimeUnit.MINUTES);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    }
    
    class KafkaConsumerRunner implements Runnable { 
        private final AtomicBoolean closed = new AtomicBoolean(false);
        private KafkaConsumer consumer;
    
        /** * 初始化配置 */
        private Properties initConfig() { 
            Properties props = new Properties();
            props.put("bootstrap.servers", "localhost:9092");
            props.put("group.id", "test-group");
            props.put("enable.auto.commit", false);
            props.put("session.timeout.ms", 30000);
            props.put("max.poll.records", 1000);
            props.put("max.poll.interval.ms", 5000);
            props.put("key.deserializer", StringDeserializer.class.getName());
            props.put("value.deserializer", StringDeserializer.class.getName());
            return props;
        }
    
        public void run() { 
            Properties pros = initConfig();
            consumer = new KafkaConsumer<String, String>(pros);
            try { 
                consumer.subscribe(Arrays.asList("flink_json_source_4"));
                while (!closed.get()) { 
                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(10000));
                    // 执行消息处理逻辑
                    for (ConsumerRecord<String, String> record : records) { 
                        System.out.printf("Context: Thread-name= %s, topic= %s partition= %s, offset= %d, key= %s,value= %s\n", Thread.currentThread().getName(), record.topic(), record.partition(), record.offset(), record.key(), record.value());
                    }
    
                }
            } catch (WakeupException e) { 
                if (!closed.get()) throw e;
            } finally { 
                consumer.close();
            }
        }
    
    
        // Shutdown hook which can be called from a separate thread
        public void shutdown() { 
            closed.set(true);
            consumer.wakeup();
        }
    }

*  wakeup主要用于唤醒polling中的consumer实例。如果你使用了多线程（即把KafkaConsumer实例用于单独的线程），你需要有能力在另一个线程中“中断”KafkaConsumer所在实例的执行。wakeup就是用这个的

输出信息：我们已经看到多个线程在消费了

![image-20210312113933444][]

这里我们思考一个问题，那就是我们并没指定那个那个线程去消费哪一个partition,按照我们上面的设计，我们应该是一个线程区去消费一个partition,从单线程的消费或者是单个consumer 的消费实例的角度，这里不应该是每个线程区消费区部分区吗？那为什么这里可以做到一个线程恰好只消费了一个分区呢。

#### 消费者程序使用单或多线程获取消息 ####

*  消费者程序使用单或多线程获取消息，同时创建多个消费线程执行消息处理逻辑。获取消息的线程可以是一个，也可以是多个，每个线程维护专属的 KafkaConsumer 实例，处理消息则交由特定的线程池来做，从而实现消息获取与消息处理的真正解耦。

![img][img 1]

##### 代码实现 #####

这个方案本质上和上面有点相似，不同的是在每个consumer 获取到数据之后，处理数据的时候是多线程的。至于获取数据的过程可以设计成单线程的也可以是多线程的。

## 总结 ##

1.  Standalone Consumer 的优势和使用场景
2.  多线程消费者的实现思路
3.  其实我们自己实现的多线程消费其实就是一种Standalone Consume的模式
4.  所以到这里我们看到kafka 总共有这么几种消费模式
    
    1.  consumer group
    2.  Standalone
    3.  自定义多线程

[image-20210312170555068]: /images/20221023/9551688b83a9412c8b6a04be0104de1e.png
[image-20210306221623211]: /images/20221023/eb064645b63f457c857362b1a758beec.png
[img]: /images/20221023/97348997a24946908fbda3ab133940e4.png
[image-20210312113933444]: /images/20221023/a8522278d0124958b18940afa0578118.png
[img 1]: /images/20221023/8942458346f54994b57990ade44b1fd5.png