扯什么kafka顺序消费,然后呢,古尔丹,代价是什么

这里的重点是有序追加到磁盘，而不是严格意义上的完全有序性。
几年前参加了一大数据岗位面试，95%的时间在扯java基础（这个可以有）和java web相关。剩下大约5%的时间换了人聊了一个kafka问题，算是大数据直接相关的东西吧。
于是有以下对话。

我：呃，我觉得不能。
几秒停顿，措词中。

M君带着一丝得意，看看我的简历。
又抬起头，仿佛在说，你改悔罢！

M：其实我们公司没有大数据开发，大数据相关用的XXX（不记得了，大约是某公司的一个什么大数据一揽子解决方案）

该公司是做车联相关的产品的（没有自己独立的大数据平台，应该车辆用户不多，数据不大，业务不复杂），
凑巧，我也刚好做过某网红新能源车相关的大数据平台。
这里结合新能源车背景来聊一聊kafka在该背景业务场景下，单分区顺序消费到底靠不靠谱。

生产端

1.终端问题

终端故障，网络或未知原因
比如车辆传感器故障等问题导致本身就乱序发送了，徒之奈何？

常规性地发现，网络情况达到小时级别的延迟。

如果实时计算，需要数据延迟尽可能的小，在watermark机制（这部份最后会提到）下，超出部份数据将不会被纳入计算。这样行程充电等业务就会被漏算，或者一个完整的过程会被切割等异常情形。

这时候作为一线开发者，如果一开始答应了产品/运维为了时效性而使用实时计算，到时候出了问题，你能用各种理由解释不是我们的问题？
当初规划选型的时候考虑到了吗？有备案吗？现在还认可吗？

2.数据倾斜

对吧？
也就是我们今天讨论的前提是基于一个常识，当我们讨论kafka能否顺序消费，一定是分区内才有讨论的可能，跨分区整个topic是不能够的。

这样kafka topic就只能有一个分区，这样的kafka集群吞吐量不敢想象。

kafka的发送分区策略：

kafkaTemplate.send(topic, info;

```
kafkaTemplate.send(topic,key, info; 
```

org.apache.kafka.clients.producer.internals.DefaultPartitioner

如果指定了自定义分区策略，不管指没指定key,以自定义策略为准。

@Component
public class DefinePartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster {
        // 这里也可以配置分区数或者定时获取分区数
        return key.hashCode( % (cluster.partitionsForTopic(topic.size( - 1;
    }

    @Override
    public void close( {
    }
    @Override
    public void configure(Map<String, ?> configs {
    }
}

然后指定分区策略
spring.kafka.producer.properties.partitioner.class = com.nyp.test.service.DefinePartitioner

但是在实践的过程当中，我们会发现，有的车作为长途或短途的运输车辆，或者作为网约车，那么每天上报的数据会相对较大，
而有的车当天没有出行或其它充电等任何操作，则没有上报数据。

可能会导致以下情况（2，3主要针对大数据框架）

吞吐下降、延迟增大
数据单点和频繁 GC 导致吞吐下降、延迟增大。
系统崩溃
严重情况下，过长的 GC 导致 TaskManager 失联，系统崩溃。

3.扩容分区的代价

简单点说，kafka的数据与broker是存放在一起的，如果要加broker，就需要将数据平衡到新的broker。
而Pulsar的架构则是节点与数据分离，消息服务层与存储层完全解耦，从而使各层可以独立扩展，所以扩容的时候会非常方便。当然这不是本文的重点。

当kafka需要扩容或者对topic增加分区时，由第2点我们得知，数据将发往哪个分区将由key%分区数决定,当分区数量变化后，所有的现有数据在进行扩容或重分区的时候都必须进行key%分区数进行重路由。

4.单分区，A，B消息顺序发送，A失败B成功，A再重试发送，变成BA顺序？

4.1 消息的发送

kafka需要在单分区保证消息按产生时间正序排列，至少应该保证按消息产生的时间正序发送。
假设消息源严格按照时间产生的前提，

同步发送，阻塞直至发送成功，返回SendResult对象，里面包含ProducerRecord和RecordMetadata对象。
SendResult result = kafkaTemplate.send(topic, key, info.get(;

异步发送，返回一个ListenableFuture对象，大家应该对Future不陌生。此对象可以添加回调方法。在成功或失败时执行相应的任务。

ListenableFuture<SendResult<Object, Object>> listenableFuture = kafkaTemplate.send(topic, key, info;
listenableFuture.addCallback(new ListenableFutureCallback<SendResult<Object, Object>>( {
    @Override
    public void onFailure(Throwable ex {

    }

    @Override
    public void onSuccess(SendResult<Object, Object> result {

    }
};

同时，异步发送需要添加相应的配置，比如一次提交多少条数据，比如如果数据迟迟没有达到发送数据量，需要设定一个最大时间，超过这个时间阀值需提交一次，等等。
注意后两个参数的配置。
不同版本之间，参数名称会有差异。

batch.size

此参数控制的发送批次的大小是以字节数，而不是数据条数。
此参数控制粒度为分区，而不是topic。当发往某个分区的数据大于等于此大小时将发起一次提交。

合理控制此参数。

linger.ms
这个设置给出了批处理延迟的上限:一旦我们获得了一个分区的batch_size值的记录，无论这个设置如何，它都会立即发送，但是如果我们为这个分区积累的字节少于这个数，我们将在指定的时间内“逗留”，等待更多的记录出现。该设置默认为0(即没有延迟。例如，设置LINGER_MS_CONFIG =5可以减少发送的请求数量，但在没有负载的情况下，发送的记录将增加5ms的延迟。
max.block.ms
前两个参数能阻塞(等待多长时间。
buffer.memory
生产者可以用来缓冲等待发送到服务器的记录的内存的总字节数。如果发送记录的速度比发送到服务器的速度快，生产者将阻塞max.block.ms，之后它将抛出异常。
这个设置应该大致对应于生产者将使用的总内存，但不是硬性限制，因为不是生产者使用的所有内存都用于缓冲。一些额外的内存将用于压缩(如果启用了压缩以及维护正在运行的请求。

4.2 消息的确认(ack

前面消息已经发送出去了，但要保证不丢消息，不重发消息，即Exactly Once 精次一次性消费，至少需要保证生产端的消息确认机制。
acks参数控制的是消息发出后，kafka集群是否需要响应，以及响应的级别。

如果设置为0，那么生产者将不会等待服务器的任何确认。该记录将立即添加到套接字缓冲区并被认为已发送。在这种情况下，不能保证服务器已经接收到记录，重试配置将不会生效(因为客户端通常不会知道任何失败。为每条记录返回的偏移量将始终设置为-1。
为方便记忆，这里的0指是的需要0个节点确认。

这将意味着leader将记录写入其本地日志，但将在不等待所有follower完全确认的情况下进行响应。在这种情况下，如果leader在确认记录后立即失败，但在follower复制它之前，那么记录将丢失。
为方便记忆，这里的1指的是只需一个节点确认，这里一个节点肯定指的是主节点leader.
这是最高级别的确认机制，同时也意味着吞吐量受到限制。它将等待leader和所有follower副本都响应，才认为发送完毕。
为方便记忆，这里的all指的是需要所有节点确认。

4.3 幂等性

答案是可以，只要开启幂等性，在Producer ID（即PID）和Sequence Number的基础上，消息最终将保持AB的顺序。

后端各微服务之间调用也有重试，也是同样的道理。

Exactly Once实际上有相同的地方，通过设置enable.idempotence=true 开启幂等性，它的基础或前提条件是，会自动设置ack=all。

enable.idempotence=true
显式开启幂等性。kafka 3.0以上的版本，此值为false，这里应该显式设置。

kafka集群的副本数至少应大于1
kafka 3.0 以后的版本，此值为1，这里应该显式设置。
在阻塞之前，客户端将在单个连接上发送的未确认请求的最大数量。请注意，如果将此设置设置为大于1，并且存在失败的发送，则存在由于重试(即，如果启用了重试而导致消息重新排序的风险。
默认值为5，如果要开启幂等性，此值应<=5。
但如果引值>1 <=5 不会报错，但还是有乱序的风险。
重试次数应大于0，如果没有重试，当A失败时不会再成功，也就无从谈起。

enable.idempotence=true，但没有显式设置3，4，5，则系统将选择合适的值。如果设置了不兼容的值，将抛出ConfigException。
同时，为保证完整性，消费端应保证 enable.auto.commit=false,isolation.level=read_committed,即自动确认改为手动确认，事务隔离级别改为读已提交。

4.3 幂等性原理

kafka为解决数据乱序和重发引入了PID和Sequence Number的概念。每个producer都会有一个producer id即PID。这对用户不可见。

同样，Broker端也会为每个<PID, Topic, Partition>维护一个序号，并且每Commit一条消息时将其对应序号递增。

如果消息序号比Broker维护的序号差值比1大，说明中间有数据尚未写入，即乱序，此时Broker拒绝该消息

发送失败后会重试，这样可以保证每个消息都被发送到broker。

A在生产端为11，B为12，
由于某种原因，A失败了，此时broker端的Sequence Number仍然为10
此时，B到达broker,它为12，大于10，且它们之间的差异大于1，此时拒绝消息B.B消息发送失败。
然后A重试，成功，Sequence Number变为11，
再然后B重试，此时成功。
最终，AB两条消息以最初的顺序写入成功。

消费端（非大数据模式）

5 单线程和多线程都不能保证跨分区顺序

先搞搞一个测试demo测试多线程消费

public void sendDocInfo(String info {
        try {
            Random random = new Random(; 
            kafkaTemplate.send("test10", random.nextInt(9+"", info + "_" + i.get(; 
        } catch (Exception e {
            log.error("kafka发送异常 " + e;
        }
    }

在消费端打印消费，带上分区ID。

@KafkaListener(
            topics = "test10",
            groupId = "heilu-group"
    
    public void handle(List<ConsumerRecord<String, String>> records, Acknowledgment ack{
        records.forEach(e -> {
            log.info(e.partition( +" 分区接收到消息 : " + e.value(;
        };
        ack.acknowledge(;
    }

可以很明显的看到跨分区乱序。

6.线程-分区一一对应

每个线程只消费一个对应的分区

@KafkaListener(
            groupId = "test-group",
            topicPartitions ={@TopicPartition(topic = "test10", partitions = { "0"}}

如图

所以没做这块的演示。

response.error不为NONE的情况下，才做canRetry判断。

7.大数据领域的解决(缓解方案watermark机制。

对于这种情况，大数据框架的共识是，对于数据乱序延迟，我们要等，但不能无限等待下去。
因此flink/spark引入了watermark俗称水印机制。

缓解数据的延迟和乱序，而不是彻底解决该问题。
就像开篇所说的第1点，车辆跑在路上总会有各种突发状态，传感器会老化，深山老林信号不好，这种情况连终端生产厂商都无法彻底解决，下游数据厂商怎么能根除呢？

如果对window不了解的，可以参考我之前写的这篇文章关于我因为flink成为spark源码贡献者这件小事

不是搞大数据的，对大数据不感兴趣的，可以跳过这一部份。

watermark = max(eventTime - 允许迟到的长度
window中，不考虑allowLateness，当watermark等于大于end-of-window时，窗口触发计算和销毁。比如：
1. 有一个窗口`[12:00-12:05`,watermark允许迟到1分钟，接收到两条数据时间分别为`12:03:43`,`12:05:23`，那么watermark = `12:05:23 - 1 minute = 12:04:23` 小于12:05,所以窗口没有结束,不触发计算
  注:严格意义来讲，[watermark = `12:05:23 - 1 minute -1ms`] 因为 end-of-window判断的时候是>=
2. 当接收到一条数据时间为12:06时，窗口触发计算如果allowLateness>0,窗口延迟销毁，假如来了一条数据时间为12:04:49会再次触发窗口计算假如来了一条数据时间为12:05:01,不会进行当前窗口，会进入到下一个窗口
考虑到代码并发度与上游(如kafka,socket分区数不匹配可能会导致有些分区消费不到数据，如测试socket只有一个分区，而flink代码中有8个并发度，那么
而flink的watermark在多并发度下，以最迟的那个为准，所以
1. 高版本里 .withIdleness(Duration.ofSeconds(x 在这个时间里，如果有空闲分区没有消费数据，那么它将不持有水印，即全局水印的推进将不考虑这些空闲分区。
如果flink任务收到一个错误数据，远超现在的系统时间，如2100-09-09 00:00:00,在除了空闲分区外的分区都收到这样的数据，那么flink任务的watermark 将超过系统时间，那么正常数据将不会被系统正常处理。这时，在watermark生成器这里要做特殊处理。
Watermark怎样生成？实时生成和周期性生成(时间或者条数，别忘了第5条。

nc -lk 9090可生产数据。

可以看下我这篇文章。

public static void main(String[] args {
        Configuration configuration = new Configuration(;
        configuration.setInteger("heartbeat.timeout", 180000;
        configuration.setInteger(RestOptions.PORT, 8082;
        StreamExecutionEnvironment streamExecutionEnvironment =
                StreamExecutionEnvironment.getExecutionEnvironment(configuration;
        // 并行度和上游数据分区数对watermark生效的影响
        // streamExecutionEnvironment.setParallelism(1;
        // nc -lk 9090
        DataStream<TestObject> dataStream =
                streamExecutionEnvironment
                        .socketTextStream(
                                "192.168.124.123",
                                9090
                        .map(
                                e -> {
                                    try {
                                        Gson gson = new GsonBuilder(
                                                .setDateFormat("yyyy-MM-dd HH:mm:ss"
                                                .create(;
                                        TestObject object = gson.fromJson(e, TestObject.class;
                                        return object;
                                    } catch (Exception exception {
                                        exception.printStackTrace(;
                                        System.out.println("异常数据 = " + e;
                                        return new TestObject(;
                                    }
                         };
        try {
            OutputTag<TestObject> lateOutput = new OutputTag<>("lateData", TypeInformation.of(TestObject.class;
            SingleOutputStreamOperator result = dataStream
                    .filter(e -> StringUtils.isNoneBlank(e.key
                    .assignTimestampsAndWatermarks(
                            (WatermarkStrategy<TestObject>
                                    WatermarkStrategy
                                   .<TestObject>
                                           forBoundedOutOfOrderness(Duration.ofSeconds(10
                                   .withTimestampAssigner(
                                           (row, ts -> {
                                               System.out.println("source = " + row;
                                               DateTimeFormatter dtf2 = DateTimeFormatter
                                                       .ofPattern("yyyy-MM-dd HH:mm:ss", Locale.CHINA;
                                               Long time = row.getTime(.getTime(;
                                               System.out.println("time = " + time;
                                               // 如果eventTime > 系统时间，这里要做处理
                                               // TODO 如果eventTime远小于系统时间，可能会拖慢整体的Watermark
                                               Long now = System.currentTimeMillis(;
                                               return
                                                       time > now ? now : time;
                                           }
                                           
                                            .withIdleness(Duration.ofSeconds(5

                    
                    .keyBy(e -> e.key
                    .window(
                            SlidingEventTimeWindows.of( Time.seconds(60 * 2, Time.seconds(60
                    // 将延迟的数据旁路输出
                    .sideOutputLateData(lateOutput
                    .process(
                            new ProcessWindowFunction<TestObject, Object, String, TimeWindow>( {
                                @Override
                                public void process(String s, Context context, Iterable<TestObject> elements, Collector<Object> out throws Exception {
                                    System.out.println("watermark = " + context.currentWatermark(;
                                    System.out.println("watermark = " + new Timestamp(context.currentWatermark(
                                            +" window.start = " + new Timestamp(context.window(.getStart(
                                            +" window.end = " + new Timestamp(context.window(.getEnd(;
                                    elements.forEach(e -> System.out.println("e + " + e;
                                }
                    };


            result.print(;
            // 迟到不处理的数据
            result.getSideOutput(lateOutput.print(;

            streamExecutionEnvironment.execute("WaterMark test";
        } catch (Exception exception {
            exception.printStackTrace(;
        }
    }


    @Data
    @NoArgsConstructor
    public static class TestObject {
        private String key;
        private Timestamp time;
        private float price;
    }

8. 小结

kafka为了吞吐量，在生产端设计了顺序追加模式，这两者才是因果。
得益于此，kafka单分区内的数据可以变得有序，这只是一个副产品。

分区节点间的数据倾斜带来的性能问题，
分区节点扩容的代价，
幂等性所需要代价带来的吞吐量限制，
以及消费端的限制。

幂等性更多的是做一次精准消费，防止重复消费，有序只是副产品。
有且只有一次精准消费，可比什么劳什子有序消费重要得多！

但它不是装X工具。

参考：

https://nightlies.apache.org/flink/flink-docs-release-1.15/zh/docs/dev/datastream/event-time/generating_watermarks/#watermark-策略和-kafka-连接器

https://juejin.cn/post/7200672322113077303
https://juejin.cn/post/7226612646543818807

编程笔记 » 扯什么kafka顺序消费,然后呢,古尔丹,代价是什么