Kafka Producer模块分析

电玩女神 2023-02-23 03:52 1阅读 0赞

# 1. 概述 #

一、Kafka Producer包含哪些部分  
其实我们讨论producer时，指的是用户接触的clients.producer这个util包，其中包含了发送数据到哪台机器、怎样序列化、分批发送与拒绝消息等等的发送策略。  
但这些讨论不包括Kafka接到这些消息后怎样处理的问题，这些是broker（即kafka server端）需要去处理的部分。  
二、Producer与Broker的关系  
producer会维护将要发送的topic表，在必要时向broker要这些topic的元信息，并在client端维护这些元信息，来决定每条消息的去向。  
从broker角度来看，它提供了元信息给producer以后，就无法控制producer的行为了。可以这么理解——broker本身是无状态的机器，主要提供的是元信息的消息读写的接口。  
三、Producer发送消息的流程简介

1.  在发送消息时，等待元信息的更新
2.  将key\\value序列化为byte\[\]，计算出数据的大小是否超出限制
3.  计算出key对应的partition，以确认将消息发往哪里
4.  使用`accumulator`将数据放到topic\\partition对应的缓冲区中
5.  使用NetworkClient类（背后是nio selector）定期的把缓冲区中的内容发出

注：本文基于Kafka 0.9版本，KafkaProducer的代码更新时间在2015-12-06

# 2. 字段与主要方法 #

## 2.1 KafkaProducer中的字段 ##

private static final AtomicInteger PRODUCER_CLIENT_ID_SEQUENCE = new AtomicInteger(1);
    
        private String clientId;                           // client.id参数：用于日志与打点，以及发请求带的参数
        private final Partitioner partitioner;        // partitioner.class参数：指定一个Partitioner类用于计算key对应的分片，可自定义
        private final int maxRequestSize;          // max.request.size参数：单条消息的内存byte数大于此值时，抛出RecordTooLargeException异常
        private final long totalMemorySize;        // buffer.memory参数：同上，但额外作为accumulator->BufferPool中的`The maximum amount of memory that this buffer pool can allocate`，似乎是accumulator中暂存消息最多可用的内存量。
        private final Metadata metadata;          // 维护了一些topic的元信息
        private final RecordAccumulator accumulator;    // 用于堆积消息，按情况batch发送
        private final Sender sender;      // 发送消息
        private final Thread ioThread;   // 把this.sender包装成一个Thread
        private final CompressionType compressionType;    //compression.type参数，压缩数据的方式，主要用在accumulator内。有none\gzip\snappy\lz4几种
        private final Metrics metrics;      // TODO 监控信息？
        private final Sensor errors;    // TODO 监控信息？
        private final Time time;          // KafkaProducer对象的创建时间
        private final Serializer<K> keySerializer;        // key序列化方法 
        private final Serializer<V> valueSerializer;    // value序列化方法
        private final ProducerConfig producerConfig;    // 外部把配置传入，但没多大用处
        private final long maxBlockTimeMs;      // 来源不明TODO，用处似乎是限制send的一些IO操作总时间（waitOnMetadata堵塞的时间，与accumulator.append中waitMemory堵塞的时间之和，要低于此值）
        private final int requestTimeoutMs;    //参数`timeout.ms` or `request.timeout.ms`，用途不明TODO

## 2.2 KafkaProducer提供的api ##

### 2.2.1 发送一条Kafka消息 ###

public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback)

1.  该方法做一定预处理后（详见第3节），将消息塞进accumulator缓冲区后，等待异步发送并返回，在消息实际发送完，即取到请求的返回值后，执行callback参数的回调方法。
2.  返回体中包装的RecordMetadata类型，带有消息最终发到的partition与offset，可用future.get()堵塞到消息发送完成。
3.  官方建议：由于callback会执行一个IO线程，因此建议处理速度足够快，否则会影响到其它线程。如果需要执行堵塞或计算量大的逻辑，建议另开一个Executor来并发处理callback来的数据。
4.  可能抛出的异常：InterruptException（堵塞时被打断）、SerializationException（序列化失败）、BufferExhaustedException（buffer空间用尽）

### 2.2.2 刷新缓存 ###

public void flush()

此方法会将accumulator缓冲区中，尚未发出的那些消息变成可发送状态。调用此方法会让当前线程堵塞住，并保证**调用flush()之前曾发送的所有消息都执行完成，即kafka服务端确实收到这些消息**。  
值得注意的是，在调用flush陷入堵塞时，其它线程仍然可以继续调用send发送消息。  
以下是官方举例的一个使用case：

// 消费100条消息，打入另一个topic后，确保所有数据发送成功，再commit消费进度
    for(ConsumerRecord<String, String> record: consumer.poll(100))
         producer.send(new ProducerRecord("my-topic", record.key(), record.value());
    producer.flush();
    consumer.commit();

### 2.2.3 topic的partition信息 ###

public List<PartitionInfo> partitionsFor(String topic)

可获取指定topic的partition元信息，若本地metadata缓存中没有此topic，则堵塞等待更新元信息。  
超时时间定为maxBlockTimeMs，堵塞可被打断抛出InterruptException。

### 2.2.4 指标信息 ###

public Map<MetricName, ? extends Metric> metrics()

获取监控信息，或说kafka使用过程中的打点。（TODO待补充）

### 2.2.5 关闭producer ###

public void close(long timeout, TimeUnit timeUnit)

关闭当前producer，堵塞当前线程等待所有sent请求完成。

## 3. 发送细节与机制详述 ##

### 3.1 获取Metadata ###

以下为Metadata中的字段

private final long refreshBackoffMs;  // 更新失败时最小的再次刷新间隔时间
        private final long metadataExpireMs;  // 过期时间, 默认60s
        private int version;  // 每次更新version自增
        private long lastRefreshMs;  // 最近的更新时的时间
        private long lastSuccessfulRefreshMs; // 最近成功更新的时间
        private Cluster cluster;  // 保存topic与partition、结点与partition等等的关系
        private boolean needUpdate; // 需要更新 metadata
        private final Map<String, Long> topics; // topic与对应的过期时间的对应关系
        private final List<Listener> listeners; // 事件监控者
        private boolean needMetadataForAllTopics; // 是否强制更新所有的 metadata

producer在发送前执行`waitOnMetadata(String topic, long maxWaitMs)`来获取元信息。此方法判断缓存中是否有topic信息，没有的话进入以下循环：

int version = metadata.requestUpdate();    // 告诉metadata信息需要更新了
                sender.wakeup();                           // 唤醒sender来更新metadata
                metadata.awaitUpdate(version, remainingWaitMs);  // 进入堵塞直到版本更新

上述的awaitUpdate操作中，metadata会堵塞住，等待sender的一系列请求成功后调用Metadata.update来唤醒自己。sender是kafka client的一个network util包装，后面再详细介绍。

如果这个过程超时了，或者topic unauthorized（TODO：想想unauthorized是什么情况？），就开始抛出异常。如果没有异常的顺利完成，topic对应的partition信息就会被producer收到并cache在内存中，producer可用此信息来发送消息。

### 3.2 消息的发送与缓冲区 ###

一、计算partition与结点  
经过3.1中讨论的元信息获取，producer就获取了这条消息topic有几个partition（分片）以及这些分片对应的结点信息。  
需要发往哪一个partition：

1.  观察消息是否指定了partition，未指定的话走partition类计算数据分片；
2.  默认partitioner类：如果消息带有key则按hash(keyBytes)%numPartitions，不带有key则按round-robin轮流发往各个partition
3.  可以通过partitioner.class配置，自定义partitioner类，按照业务需求来自定义partition计算方式。  
    确认了消息发往哪一个partition，也就能从元信息中找到partition对应的leader结点了。

二、accumulator提供的消息缓冲  
确认了消息要发送的结点后，下一步就是将消息内容与结点信息塞进accumulator。  
accumulator是一个buffer，负责把producer发送的消息累积在那里，直到特定条件时发出。其buffer采用一个map<TopicPartition, Deque >来存储，每一个topic\\partition下都是一个RecordBatch队列。  
队列中每一个RecordBatch元素，都代表一组kafka消息。在一个RecordBatch满了，或者其它线程调用flush()\\close()等操作时，producer都会唤醒sender，来负责把buffer中的RecordBatch发到kafka服务端去。

### 3.3 消息缓冲的flush机制 ###

在accumulator中使用了一个`AtomicInteger flushesInProgress`来表示是否在flush状态。  
在Sender类（在4.2中会详细介绍Sender类）每次迭代时，会先算出现在**要往哪些结点发送消息**。计算逻辑是一个缓冲区满了、超时、没空间，或当前在flush状态，就会认为消息需要发出去了。  
因此：当我们将flushesInProgress加一，则会在Sender下一轮迭代时，将所有待发数据的结点全标成待发送的。flush()流程示意如下：

1.  将flushesInProgress++
2.  将sender唤醒去干活
3.  等待所有未完成的缓冲消息发送完成
4.  将flushesInProgress–

因此我们可以理解为，flush()操作会将当前所有缓存都发出一次。这就保证了在flush()前所有消息都会被发送完成，flush状态才结束；但保持flush状态的过程中新加入的消息，我们无法确定其状态。

## 4. 拓展：Sender ##

前面提到更新metadata，或发送accumulator中的批量消息，以及flush()，此类网络I/O操作，都会执行一个相同的代码`sender.wakeup()`。所以这个万能的sender到底是什么东西呢？

### 4.1 初始化与启动 ###

下面是producer的构造器中，把Sender初始化出来的地方：

this.sender = new Sender(client,
                        this.metadata,
                        this.accumulator,
                        config.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG),
                        (short) parseAcks(config.getString(ProducerConfig.ACKS_CONFIG)),
                        config.getInt(ProducerConfig.RETRIES_CONFIG),
                        this.metrics,
                        new SystemTime(),
                        clientId,
                        this.requestTimeoutMs);
                String ioThreadName = "kafka-producer-network-thread" + (clientId.length() > 0 ? " | " + clientId : "");
                this.ioThread = new KafkaThread(ioThreadName, this.sender, true);
                this.ioThread.start();

producer构造了一个Runnable的sender的成员变量，并开启线程执行。

### 4.2 Sender做了什么 ###

其主要工作是不断的循环以下操作：

1.  通过accumulator中可发出的数据包，以及metadata信息，算出现在准备发送消息的结点。（若找不到leader信息，标记一下更新metadata）
2.  "倒出"accumulator中的数据（把超时未发的数据打个记录），把数据包和对应结点打包创建出ClientRequest。
3.  把这些请求塞进sender中维护的client（使用nio维护的请求队列），并通过`client.poll`来发送出去。（TODO 网络相关的操作补充）
4.  poll()的时候会执行一个`metadataUpdater.maybeUpdate(now)`的操作：
    
     *  先计算出下次刷新时间：根据是否请求刷新元数据（见本节第1条及3.1节中的刷新策略），或`metadata.max.age.ms`配置中的最久没有成功刷新的时间，并考虑到`retry.backoff.ms`配置中的最短刷新间隔，得出一个下次刷新的时间。
     *  考虑timeToNextReconnectAttempt，以及是否正在获取metadata信息，来更新下次刷新时间。
     *  如果已经到了可刷新时间点，获取leastLoadedNode，并将打向这个node的请求塞进client，也就搭上了这一波client.poll网络请求的车。

可以举个例子帮助解释：

1.  第一轮迭代中，如果需要某些node A的元信息并发现缺失了，此时sender会请求更新metadata。那么在当前这一轮迭代结束的poll(now)操作中，通过metadataUpdater.mayBeUpdate(now)发现需要更新metadata，并发出更新的网络请求。
2.  第二轮迭代中：由于metadata的更新请求只发出没处理，因此依然没有缺失node的信息。但是在这一轮迭代结束的poll()中，如果metadata请求返回了，就可以通过返回值补上node A的信息。
3.  第三轮迭代中：可正常获取node A的信息，并向A发消息了。

## 5. 其它 ##

TODO，待补充kafka 0.9到现在的更新内容