浅谈MQ-kafka生产者原理

柔光的暖阳◎ 2023-01-21 10:26 104阅读 0赞

# Kafka生产者原理 #

## 1.生产者消费发送流程 ##

消息发送的整体流程。生产端主要由两个线程协调运行。这两条线程分别为main线程和sender线程（发送线程）。

![image-20210505165539388][]

### 1.1拦截器 ###

拦截器的作用是实现消息的定制化（类似于：Spring Interceptor、Mybatis的插件、Quartz的监听器）。

这个拦截器是在那里定义的呢？

List<String> interceptors = new ArrayList<>();
    interceptors.add("自定义的拦截器类全路径")
    props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);

可以在生产者属性中添加多个拦截器，形成拦截器链。

举个例子，按量付费的实现

public class ChargingInterceptors implementes ProducerInterceptors<String, String> { 
      @Override
      public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) { 
        println("1分钱 1条消息");
        return record;
      }
      @Override
      public void onAcknowledgement(RecordMetadata metadata，Exception exception) { 
        println("消息被服务端接受了");
      }
      @Override
      public void close() { 
        println("生产者关闭");
      }
      @Override
      public void configure(Map<String, ?> configs) { 
        println("configure...");
      }
    }

### 1.2序列化 ###

kafka针对不同的数据类型自带了相应的序列化工具（他们都继承Serializer.java接口）：

*  ByteArraySerializer
 *  ByteBufferSerializer
 *  DoubleSerializer
 *  FloatSerializer
 *  IntegerSerializer
 *  LongSerializer
 *  ShortSerializer
 *  StringSerializer
 *  UUIDSerializer

除了自带的序列化工具之外，可以使用如Avro、JSON、Thrift、Protobuf等，或者使用自定义类型的序列化器来实现，实现Serializer接口即可。

### 1.3路由指定（分区器） ###

路由指定有四种情况：

1.  指定了partition。
    
    直接将指定的值作为partition值。
2.  没有指定partition，自定义了分区器。
    
    自定义分区器，将使用自定义的分区器算法选择分区，比如SimplePartitioner，用ProducerAutoPartition指定，发送消息。
    
        props.put("partitioner.class", "com.xxx.xxx.partition.SimplePartition");
3.  没有指定partition，没有自定义分区器，但是key不为空。
    
    这种情况下，使用默认分区器DefaultPartitioner，将key的hash值与topic的Partition数进行取余得到Partition值；
4.  没有指定partition，没有自定义分区器，但是key是空的。
    
    这种情况下，第一次调用时随机生成一个整数（后面每次调用在这个整数上自增），将这个值与topic可用的Partition总数取余得到Partition值，也就是常说的轮训算法。

### 1.4消息累加器 ###

选择分区以后并没有直接发送消息，而是把消息放入消息累加器。

RecordAccumulator.RecordAppendResult result = accumulator.append(tp, timestamp, serializedKey, serializedValue, headers, interceptCallback, remainingWaitMs);

RecordAccumulator本质上是一个ConcurrentMap：

ConcurrentMap<TopicPartition, Deque<ProducerBatch>> batches;

一个partition一个Batch。batch满了之后，会唤醒Sender线程，发送消息。

## 2.服务端响应ACK ##

### 2.1服务端响应策略 ###

生产者的消息是不是发出去就完事了?如果说网络出了问题，或者说kafka服务端接收的时候出了问题，这个消息发送失败了，生产者是不知道的。

所以，kafka服务端应该要有一种响应客户端的方式，只有在服务端确认以后，生产者才发送下一轮的消息，否则重新发送数据。

服务端什么时候才算接收成功呢?因为消息是存储在不同的partition里面的，所以是写入到partition之后响应生产者。

![image-20210505173057088][]

当然，单个Partition（leader）写入成功，还是不可靠，如果多个副本，follower也要写入成功才可以。

服务端发送ACK给生产者总体上有两种思路：

第一种是需要有半数以上的follower节点完成同步，这样的话客户端等待的时间就短一些，延迟低(为什么通常来说我们部署节点都是奇数?)。

第二种需要所有的follower全部完成同步，才发送ACK给客户端，延迟相对来说高一些，但是节点挂掉的影响相对来说小一些，因为所有的节点数据都是完整的。

Kafka会选择哪种方案呢?

Kafka选择了第二种方案。部署同样机器数量的情况下，第二种方案的可靠性更高。例如部署5台机器，那么第一种方案最多可能会有2台机器丢失数据，第二种方案都不会丢失。而且网络延迟对kafka的影响不大。

### 2.2 ISR ###

如果直接采用第二种思路，不考虑网络延迟，有没有别的问题呢?

假设leader收到数据，所有follower都开始同步数据，但是有一个follower出了问题，没有办法从leader同步数据。按照这个规则，leader就要一致等待，无法发送ack，可以说成为了害群之马。

打个比方，每天早上皇帝都要把所有的太子召集在一起，开个早会，传达一下信息。如果有哪位太子没有来，这个传达信息的早会就开不成。但是某一天有个太子忘记调闹钟了，没起得来。所有的人都在等他。

从概率的角度来讲，这种问题肯定是会出现的，就是某个follower出问题了，怎么解决呢?

所以我们的规则就不能那么粗暴了，把规则改一下，不是所有的follower都有权利让我等待，而是只有那些正常工作的follower同步数据的时候我才会等待。

我们应该把那些正常和leader保持同步的replica维护起来，放到一个动态set里面，这个就叫做in-sync replica set(IS R) 。现在只要IS R里面的follower同步完数据之后，我就给客户端发送ACK。

对于经常性迟到，睡觉还关机的太子，看来他不关心国事，也不能指望他了，把他从太子早会微信群移除了。

如果一个follower长时间不同步数据，就要从ISR剔除。

但是这个太子就是偶然一次忘记调闹钟，就被踢出微信群，太无情了吧?比如前一天晚上它太操劳了呢?每个人都有犯错的时候嘛。

所以，到底多久没有向leader同步数据，才会被踢出ISR呢?

由参数replica.lag.time.max.ms决定(默认值30秒) 。

相当于给了你很多机会啦，如果你连续30天没来参加早会，拜拜了。当然，如果有一天你改过自新了，还可以再邀请你进入微信群加入ISR。

但是，如果有一天这个没参加早会的不是别人，而是皇帝呢?打个比方哈，皇帝驾崩了，大家还怎么开会?谁来传达信息?

皇帝没了，所有这些参加早会的人就有机会了。新一代的皇帝就要在这些人里面产生了。

所以，如果leader挂了，就会从ISR重新选举leader。

### 2.3 ACK应答机制 ###

当然，如果所有的数据都一视同仁，而且这种策略只能由服务端决定，这就不是很灵活了。有一些数据丢了无所谓，我只想要快，不管它落没落盘同没同步，怎么办呢?

Kafka为客户端提供了三种可靠性级别，用户根据对可靠性和延迟的要求进行权衡，选择相应的配置。

acks参数配置：

porps.put("acks", 1);

举例：topic的partition0有三个副本。

*  acks=0：producer不等待broker的ack，这一操作提供了一个最低的延迟，broker一接收到还没写入磁盘就已经返回，当broker故障时有可能丢失数据；
 *  acks=1 (默认)：producer等待broker的ack，partition的leader落盘成功后返回ack，如果在follower同步成功之前leader故障，那么就会丢失数据。
 *  acks=-1 (all)： producer等待broker的ack，partition的leader和follower全部落盘成功后才返回ack。
    
    acks=-1这种方案还不是很完美，如果在follower同步完成后，broker发送ack之前，leader发生故障，没有给生产者发送ack，那么会造成数据重复。
    
    在这种情况下，吧reties设置成0（不重发），才不会重复。

三种机制，性能依次递减（producer吞吐量降低），数据健壮性则依次递增。

[image-20210505165539388]: /images/20221020/8d2ab0c335ed4725bb64c92f38f0c5de.png
[image-20210505173057088]: /images/20221020/a81d89756a884e2a8b90d4ab8e01402b.png