kafka消息顺序
我们知道，kafka是一个高性能、分布式容错的消息发布-订阅系统，现在kafka也能处理流数据了。多个生产者在往kafka发送数据的时候，消息的顺序是不能保障的，也就是无序的。
有些场景，比如在用maxwell解析mysql的binlog日志的时候，发送到kafka，后面解析，通常会用jdbc的方式，将数据保存到其它系统，比如hive或者hbase等。这时候，消息的顺序是很重要的，对于一条数据，应该是先有insert，再有updae / delete，顺序反了就不行。
kafka的顺序，包括全局顺序与局部顺序。
全局顺序
全局顺序就目前的应用范围来讲，可以列举出来的也就限于binlog日志传输，如mysql binlog日志传输要求全局的顺序，不能有任何的乱序。这种的解决办法通常是最为保守的方式：
- 全局使用一个生产者
- 全局使用一个消费者（并严格到一个消费线程）
- 全局使用一个分区（当然不同的表可以使用不同的分区或者topic实现隔离与扩展）
局部顺序
其实在大部分业务场景下，只需要保证消息局部有序即可，什么是局部有序？局部有序是指在某个业务功能场景下保证消息的发送和接收顺序是一致的。如：订单场景，要求订单的创建、付款、发货、收货、完成消息在同一订单下是有序发生的，即消费者在接收消息时需要保证在接收到订单发货前一定收到了订单创建和付款消息。
针对这种场景的处理思路是：针对部分消息有序（message.key相同的message要保证消费顺序）场景，可以在producer往kafka插入数据时控制，同一key分发到同一partition上面。因为每个partition是固定分配给某个消费者线程进行消费的，所以对于在同一个分区的消息来说，是严格有序的（在kafka 0.10.x以前的版本中，kafka因消费者重启或者宕机可能会导致分区的重新分配消费，可能会导致乱序的发生，0.10.x版本进行了优化，减少重新分配的可能性）。
注意事项
消息重试对顺序消息的影响
对于一个有着先后顺序的消息A、B，正常情况下应该是A先发送完成后再发送B，但是在异常情况下，在A发送失败的情况下，B发送成功，而A由于重试机制在B发送完成之后重试发送成功了。
这时对于本身顺序为AB的消息顺序变成了BA
消息producer发送逻辑的控制
消息producer在发送消息的时候，对于同一个broker连接是存在多个未确认的消息在同时发送的，也就是存在上面场景说到的情况，虽然A和B消息是顺序的，但是由于存在未知的确认关系，有可能存在A发送失败，B发送成功，A需要重试的时候顺序关系就变成了BA。简之一句就是在发送B时A的发送状态是未知的。
针对以上的问题，严格的顺序消费还需要以下参数支持：max.in.flight.requests.per.connection
这个参数官方文档的解释是：
The maximum number of unacknowledged requests the client will send on a single connection before blocking. Note that if this setting is set to be greater than 1 and there are failed sends, there is a risk of message re-ordering due to retries (i.e., if retries are enabled).
大体意思是：
在发送阻塞前对于每个连接，正在发送但是发送状态未知的最大消息数量。如果设置大于1，那么就有可能存在有发送失败的情况下，因为重试发送导致的消息乱序问题。
所以我们应该将其设置为1，保证在后一条消息发送前，前一条的消息状态已经是可知的。
kafka消息重复
kafka生产者在发送数据的时候，通常会有同步与异步发送，异步就是缓存部分数据，达到一定条数或时间后批量发送，效率高效。那么，不管同步还是异步，消息是否发送成功，Kafka通过acks这个参数来控制的：
0---表示不进行消息接收是否成功的确认；
1---表示当Leader接收成功时确认；
-1---表示Leader和Follower都接收成功时确认；
通常为了兼顾效率与数据安全，将acks设置为1，只让每个分区的leader确认收到消息即可，不能副本是否同步数据完毕。
那么，在生产者发送数据到kafka后，如果返回成功的时候，由于网络等原因出现异常，那么生产者是收不到成功信号的，会重发，导致消息重复；消费者在成功消费后，可能还没有来得及提交偏移量，程序异常，即偏移量没有成功提交，下次启动也会重复消费。
生产者重复发送数据，消费者重复消费数据，这些都导致消息重复，那么避免重复也应该在消息的生产与消费来避免。
对于生产端：
-
每个分区使用一个单独的写入器，每当你发现一个网络错误，检查该分区中的最后一条消息，以查看您的最后一次写入是否成功
-
在消息中包含一个主键(UUID或其他)，并在用户中进行反复制
对于消费端：
- 采用exactly-once语义，消息消费结果保存与手动提交偏移量做成一个事务，比如一条sql语句既保存结果也保存偏移量，要么一起成功，要么一起失败。
- 也可以根据数据唯一字段进行重复判断
kafka时间轮
Kafka中存在大量的延迟操作，比如延迟生产、延迟拉取以及延迟删除等。Kafka并没有使用JDK自带的Timer或者DelayQueue来实现延迟的功能，而是基于时间轮自定义了一个用于实现延迟功能的定时器（SystemTimer）。JDK的Timer和DelayQueue插入和删除操作的平均时间复杂度为O(nlog(n))，并不能满足Kafka的高性能要求，而基于时间轮可以将插入和删除操作的时间复杂度都降为O(1)。时间轮的应用并非Kafka独有，其应用场景还有很多，在Netty、Akka、Quartz、Zookeeper等组件中都存在时间轮的踪影。
参考下图，Kafka中的时间轮（TimingWheel）是一个存储定时任务的环形队列，底层采用数组实现，数组中的每个元素可以存放一个定时任务列表（TimerTaskList）。TimerTaskList是一个环形的双向链表，链表中的每一项表示的都是定时任务项（TimerTaskEntry），其中封装了真正的定时任务TimerTask。
时间复杂度：O(1)
时间轮由多个时间格组成，每个时间格代表当前时间轮的基本时间跨度（tickMs）。时间轮的时间格个数是固定的，可用wheelSize来表示，那么整个时间轮的总体时间跨度（interval）可以通过公式 tickMs × wheelSize计算得出。时间轮还有一个表盘指针（currentTime），用来表示时间轮当前所处的时间，currentTime是tickMs的整数倍。currentTime可以将整个时间轮划分为到期部分和未到期部分，currentTime当前指向的时间格也属于到期部分，表示刚好到期，需要处理此时间格所对应的TimerTaskList的所有任务。
若时间轮的tickMs=1ms，wheelSize=20，那么可以计算得出interval为20ms。初始情况下表盘指针currentTime指向时间格0，此时有一个定时为2ms的任务插入进来会存放到时间格为2的TimerTaskList中。随着时间的不断推移，指针currentTime不断向前推进，过了2ms之后，当到达时间格2时，就需要将时间格2所对应的TimeTaskList中的任务做相应的到期操作。此时若又有一个定时为8ms的任务插入进来，则会存放到时间格10中，currentTime再过8ms后会指向时间格10。如果同时有一个定时为19ms的任务插入进来怎么办？新来的TimerTaskEntry会复用原来的TimerTaskList，所以它会插入到原本已经到期的时间格1中。总之，整个时间轮的总体跨度是不变的，随着指针currentTime的不断推进，当前时间轮所能处理的时间段也在不断后移，总体时间范围在currentTime和currentTime+interval之间。
如果此时有个定时为350ms的任务该如何处理？直接扩充wheelSize的大小么？Kafka中不乏几万甚至几十万毫秒的定时任务，这个wheelSize的扩充没有底线，就算将所有的定时任务的到期时间都设定一个上限，比如100万毫秒，那么这个wheelSize为100万毫秒的时间轮不仅占用很大的内存空间，而且效率也会拉低。Kafka为此引入了层级时间轮的概念，当任务的到期时间超过了当前时间轮所表示的时间范围时，就会尝试添加到上层时间轮中。
对于之前所说的350ms的定时任务，显然第一层时间轮不能满足条件，所以就升级到第二层时间轮中，最终被插入到第二层时间轮中时间格17所对应的TimerTaskList中。如果此时又有一个定时为450ms的任务，那么显然第二层时间轮也无法满足条件，所以又升级到第三层时间轮中，最终被插入到第三层时间轮中时间格1的TimerTaskList中。注意到在到期时间在[400ms,800ms)区间的多个任务（比如446ms、455ms以及473ms的定时任务）都会被放入到第三层时间轮的时间格1中，时间格1对应的TimerTaskList的超时时间为400ms。随着时间的流逝，当次TimerTaskList到期之时，原本定时为450ms的任务还剩下50ms的时间，还不能执行这个任务的到期操作。这里就有一个时间轮降级的操作，会将这个剩余时间为50ms的定时任务重新提交到层级时间轮中，此时第一层时间轮的总体时间跨度不够，而第二层足够，所以该任务被放到第二层时间轮到期时间为[40ms,60ms)的时间格中。再经历了40ms之后，此时这个任务又被“察觉”到，不过还剩余10ms，还是不能立即执行到期操作。所以还要再有一次时间轮的降级，此任务被添加到第一层时间轮到期时间为[10ms,11ms)的时间格中，之后再经历10ms后，此任务真正到期，最终执行相应的到期操作。
设计，其本源于生活。我们常见的钟表就是一种具有三层结构的时间轮，第一层时间轮tickMs=1ms, wheelSize=60，interval=1min，此为秒钟；第二层tickMs=1min，wheelSize=60，interval=1hour，此为分钟；第三层tickMs=1hour，wheelSize为12，interval为12hours，此为时钟。
在Kafka中第一层时间轮的参数同上面的案例一样：tickMs=1ms, wheelSize=20, interval=20ms，各个层级的wheelSize也固定为20，所以各个层级的tickMs和interval也可以相应的推算出来。Kafka在具体实现时间轮TimingWheel时还有一些小细节：
- TimingWheel在创建的时候以当前系统时间为第一层时间轮的起始时间(startMs)，这里的当前系统时间并没有简单的调用System.currentTimeMillis()，而是调用了Time.SYSTEM.hiResClockMs，这是因为currentTimeMillis()方法的时间精度依赖于操作系统的具体实现，有些操作系统下并不能达到毫秒级的精度，而Time.SYSTEM.hiResClockMs实质上是采用了System.nanoTime()/1_000_000来将精度调整到毫秒级。也有其他的某些骚操作可以实现毫秒级的精度，但是笔者并不推荐，System.nanoTime()/1_000_000是最有效的方法。（如对此有想法，可在留言区探讨。）
- TimingWheel中的每个双向环形链表TimerTaskList都会有一个哨兵节点（sentinel），引入哨兵节点可以简化边界条件。哨兵节点也称为哑元节点（dummy node），它是一个附加的链表节点，该节点作为第一个节点，它的值域中并不存储任何东西，只是为了操作的方便而引入的。如果一个链表有哨兵节点的话，那么线性表的第一个元素应该是链表的第二个节点。
- 除了第一层时间轮，其余高层时间轮的起始时间(startMs)都设置为创建此层时间轮时前面第一轮的currentTime。每一层的currentTime都必须是tickMs的整数倍，如果不满足则会将currentTime修剪为tickMs的整数倍，以此与时间轮中的时间格的到期时间范围对应起来。修剪方法为：currentTime = startMs - (startMs % tickMs)。currentTime会随着时间推移而推荐，但是不会改变为tickMs的整数倍的既定事实。若某一时刻的时间为timeMs，那么此时时间轮的currentTime = timeMs - (timeMs % tickMs)，时间每推进一次，每个层级的时间轮的currentTime都会依据此公式推进。
- Kafka中的定时器只需持有TimingWheel的第一层时间轮的引用，并不会直接持有其他高层的时间轮，但是每一层时间轮都会有一个引用（overflowWheel）指向更高一层的应用，以此层级调用而可以实现定时器间接持有各个层级时间轮的引用。
关于时间轮的细节就描述到这里，各个组件中时间轮的实现大同小异。读者读到这里是否会好奇文中一直描述的一个情景——“随着时间的流逝”或者“随着时间的推移”，那么在Kafka中到底是怎么推进时间的呢？类似采用JDK中的scheduleAtFixedRate来每秒推进时间轮？显然这样并不合理，TimingWheel也失去了大部分意义。
Kafka中的定时器借助了JDK中的DelayQueue来协助推进时间轮。具体做法是对于每个使用到的TimerTaskList都会加入到DelayQueue中，“每个使用到的TimerTaskList”特指有非哨兵节点的定时任务项TimerTaskEntry的TimerTaskList。DelayQueue会根据TimerTaskList对应的超时时间expiration来排序，最短expiration的TimerTaskList会被排在DelayQueue的队头。Kafka中会有一个线程来获取DelayQueue中的到期的任务列表，有意思的是这个线程所对应的名称叫做“ExpiredOperationReaper”，可以直译为“过期操作收割机”，和“SkimpyOffsetMap”有的一拼。当“收割机”线程获取到DelayQueue中的超时的任务列表TimerTaskList之后，既可以根据TimerTaskList的expiration来推进时间轮的时间，也可以就获取到的TimerTaskList执行相应的操作，对立面的TimerTaskEntry该执行过期操作的就执行过期操作，该降级时间轮的就降级时间轮。
读者读到这里或许又非常的困惑，文章开头明确指明的DelayQueue不适合Kafka这种高性能要求的定时任务，为何这里还要引入DelayQueue呢？注意对于定时任务项TimerTaskEntry插入和删除操作而言，TimingWheel时间复杂度为O(1)，性能高出DelayQueue很多，如果直接将TimerTaskEntry插入DelayQueue中，那么性能显然难以支撑。就算我们根据一定的规则将若干TimerTaskEntry划分到TimerTaskList这个组中，然后再将TimerTaskList插入到DelayQueue中，试想下如果这个TimerTaskList中又要多添加一个TimerTaskEntry该如何处理？对于DelayQueue而言，这类操作显然变得力不从心。
分析到这里可以发现，Kafka中的TimingWheel专门用来执行插入和删除TimerTaskEntry的操作，而DelayQueue专门负责时间推进的任务。再试想一下，DelayQueue中的第一个超时任务列表的expiration为200ms，第二个超时任务为840ms，这里获取DelayQueue的队头只需要O(1)的时间复杂度。如果采用每秒定时推进，那么获取到第一个超时的任务列表时执行的200次推进中有199次属于“空推进”，而获取到第二个超时任务时有需要执行639次“空推进”，这样会无故空耗机器的性能资源，这里采用DelayQueue来辅助以少量空间换时间，从而做到了“精准推进”。Kafka中的定时器真可谓是“知人善用”，用TimingWheel做最擅长的任务添加和删除操作，而用DelayQueue做最擅长的时间推进工作，相辅相成。
参考链接：