Kafka之精确一次处理语义

Dear 丶 2022-11-20 09:56 156阅读 0赞

### 文章目录 ###

*  [精确一次处理语义][Link 1]
 *  [幂等性实现原理][Link 2]

# 精确一次处理语义 #

1.  消息处理语义
    
     *  最多一次(at most once):消息可能丢失也可能被处理，但最多只会处理一次
     *  至少一次(at least once):消息不会丢失，但可能被处理多次
     *  精确一次(exactly once):消息被处理且只会被处理一次
2.  producer：
    
     *  在`0.11.0.0`之前，kafka producer默认提供至少一次（at least once）语义（重试机制）
     *  kafka在`0.11.0.0`开始引入producer精确一次性语义EOS(exactly-once semantics)。即瞬时发送错误导致的producer重试，在broker端这条消息只会被写入日志一次。如果要启用幂等性producer以及获取EOS语义，需要显示设置producer参数`enable.idempotence=true`
3.  consumer：
    
     *  最多一次(at most once)： consumer首先获取消息，然后提交offset，之后再处理消息。当提交offset后，consumer崩溃，消息可能永远不会被处理，即消息丢失
     *  至少一次(at least once)：consumer先获取消息，然后处理消息，最后提交offset。当消息处理完，consumer崩溃，此时可能会重复处理
4.  案例
    
        @Test
        public void testIdempotence() {
            Properties props = new Properties();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-master:9092,kafka-slave1:9093,kafka-slave2:9094");
            props.put(ProducerConfig.ACKS_CONFIG, "all");
            props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
            props.put(ProducerConfig.RETRIES_CONFIG, "10");
            //开启幂等性
            props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
            props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, "1000");
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        
            KafkaProducer<Integer, String> producer = new KafkaProducer<>(props);
            String topic = "testTopic";
            for (int i = 0; i < 100; i++) {
                String messageStr = "hello world " + i;
                ProducerRecord producerRecord = new ProducerRecord(topic, null, messageStr);
                producer.send(producerRecord, (recordMetadata, e) -> {
                    if (e != null) {
                        if (e instanceof RetriableException) {
                            //处理可重试瞬时异常
                        } else {
                            //处理不可重试瞬时异常
                            logger.error(e.getMessage(), e);
                        }
                    }
                    if (recordMetadata != null) {
                        logger.info(ToStringBuilder.reflectionToString(recordMetadata));
                    }
                });
            }
            producer.close();
        }

# 幂等性实现原理 #

1.  实现原理
    
     *  Producer发送到broker端的每批消息都会被赋予一个序列号用于去重，从0开始严格单调递增。
     *  每个Producer实例都有一个producer ID，producer在初始化时必须分配一个producer ID（每个topic每个分区都有自己的序列号）
     *  broker会将序列号与消息一起保存在日志中。这样即使leader副本挂掉，新选择的leader也能执行消息去重
     *  如果发送消息的序列号小于或者等于broker端保存的序列号，那么broker会拒绝这条消息的写入操作
2.  kafka以上的设计确保了即使出现重试操作，每条消息也仅仅在日志中保存一次。不过由于每个新的producer实例都会有不同的producer ID，同一个分区下序列化是递增的，所以**只能保证单个producer实例的EOS语义，无法保证多个producer实例一起提供EOS语义**，只能保证单个topic 分区的幂等性

[Link 1]: https://blog.csdn.net/usagoole/article/details/103605960#_5
[Link 2]: https://blog.csdn.net/usagoole/article/details/103605960#_64