Flink消费Kafka如何保证相同标识消息的有序性

1.需求

在某些情况下，我们需要保证flink在消费kafka时，对于某些具有相同标识的消息，要保证其顺序性。
比如说具有相同uuid的用户行为消息，要保证其消息的顺序性，这样才能有效分析其用户行为。
问题：
kafka只能保证同一个partition内的消息是顺序性的，但是整个topic下并不能保证是顺序的，那么该如何解决呢？

2.解决方案

<1> 在生产消息时，就将具有相同uuid的消息分配到同一个分区中。
扩展：kafka topic消息分配partition规则
源码：

public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
         List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
         int numPartitions = partitions.size();
         if (keyBytes == null) {
             int nextValue = counter.getAndIncrement();
             List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
             if (availablePartitions.size() > 0) {
                 int part = Utils.toPositive(nextValue) % availablePartitions.size();
                  return availablePartitions.get(part).partition();
             } else {
                 // no partitions are available, give a non-available partition
                 return Utils.toPositive(nextValue) % numPartitions;
             }
         } else {
             // hash the keyBytes to choose a partition
             return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
        }
     }

通过源码，分区器就会根据消息里面的分区参数key值将消息分到对应的partition。
1）如果没有指定key值并且可用分区个数大于0时，在就可用分区中做轮询决定改消息分配到哪个partition
2）如果没有指定key值并且没有可用分区时，在所有分区中轮询决定改消息分配到哪个partition
3）如果指定key值，对key做hash分配到指定的partition

相关实现方案链接：https://blog.csdn.net/justclimbing/article/details/79613900
实现方案1：自定义在flume拦截器中使用 kafka producer
实现方案2：使用kafka-sink

<2> 如果只是某一窗口内的统计，可以针对窗口进行时间戳排序
继承抽象类ProcessWindowFunction，
ProcessWindowFunction<IN, OUT, KEY, W extends Window>

该类一次性迭代整个窗口里的所有元素，比较重要的一个对象是Context，可以获取到事件和状态信息，这样我们就可以实现更加灵活的控制，这实际上是process的主要特点吧。该算子会浪费很多性能吧，主要原因是不增量计算，要缓存整个窗口然后再去处理，所以要设计好内存。

WindowedStream调用process方法，传入ProcessWindowFunction参数。

public class EventSessionProcessFunction extends ProcessWindowFunction<EventBase, EventBase, String, TimeWindow> {

    @Override
    public void process(String key, Context context, Iterable<EventBase> elements, Collector<EventBase> out) throws Exception {
        List<EventBase> sortedEvents = Ordering.from((Comparator<EventBase>) (x, y) -> {
            if (x.getTimestamp() - y.getTimestamp() > 0) {
                return 1;
            } else if (x.getTimestamp() - y.getTimestamp() < 0) {
                return -1;
            } else {
                return 0;
            }
        }).sortedCopy(elements);

        for (EventBase event : sortedEvents) {
            out.collect(event);
        }
    }
}