時序數據流通過Kafka隊列時可能產生的亂序緣由和解決方法

Kafka做爲一個流行的消息隊列，以分佈式高性能，高可靠性等特色已經在多種場景下普遍使用。在工業互聯網、物聯網時序數據存儲的解決方案中也有大量用到。

但在實際部署過程當中，可能會由於配置緣由致使通過Kafka的數據在接收方產生亂序，給後續處理環節帶來排序等工做，形成沒必要要的處理開銷，下降系統的處理性能和額外排序的工做。

其實能夠經過合理的規劃設計Kafka的配置和方法來避免消息在經過Kafka後亂序的產生，只須要遵循如下原則便可：

對於須要確保順序的一條消息流，發送到同一個partition上去

Kafka能夠在一個topic下設置多個partition來實現分佈式和負載均衡，由同一consumer group下的不一樣consumer去消費；這樣的機制可以支持多線程分佈式的處理，帶來高性能，但也帶來了同一消息流走了不一樣路徑的可能性，若是沒有針對性的規劃，從架構上就沒法保證消息的順序。以下圖所示，對於同一個topic的一條消息流，寫入不一樣的partition，就會產生多條路徑。

爲了確保一條消息流的數據可以嚴格按照時間順序被消費，則必須遵循一條路徑的原則，這樣才能實現FIFO（First In First Out)。

根據Kafka的文檔描述，把哪條記錄發到哪一個partition，是由producer負責：

Producers

Producers publish data to the topics of their choice. The producer is responsible for choosing which record to assign to which partition within the topic. This can be done in a round-robin fashion simply to balance load or it can be done according to some semantic partition function (say based on some key in the record). More on the use of partitioning in a second!

可見，Kafka已考慮到了確保消息順序的需求，提供了接口來實現根據指定的key值發送到同一partition的方法。 能夠看看Kafka相關源碼：

class DefaultPartitioner(props: VerifiableProperties = null) extends Partitioner {
  private val random = new java.util.Random

  def partition(key: Any, numPartitions: Int): Int = {
    Utils.abs(key.hashCode) % numPartitions
  }
}
複製代碼

從源碼上來看，Kafka支持經過Key的hash值對partition的數量求餘來實現基於Key的分配partition方法。所以咱們只要對不一樣時序消息流，找到他們不一樣的key，而且這個key是不會發生變化的，那麼就能在發送到Kafka的時候，確保每一條消息流發送到同一個partition，走惟一的路徑。所以咱們能夠經過指定Key的方式，來實現這種嚴格的時序關係。

具體實現方法

在TDengine的應用場景下，咱們一般會把某一類設備（超級表）劃分爲一個topic。對於每一個設備，會單獨建表，一個設備產生的數據，會只放到一張表裏。對於設備產生的原始數據，就須要在這個數據中找一個可以表明這個數據的ID，並且不會發生變化的字段，做爲Key值，在發送給Kafka時，帶上這個Key值。這樣就能確保該設備的全部數據流通過Kafka時，走惟一的路徑。這個ID或key每每是設備具備惟一性的設備編碼，這個編碼不只能夠做爲Kafka的Key, 也能夠做爲TDengine裏的表名。

具體實現很是簡單，在producer發送數據時，選擇一個key，經過KeyedMessage方法生成消息，而後send。以Java爲例，其餘語言能夠從Kafka文檔中找到相同功能的接口：

producer.send(new KeyedMessage<String, String>(topic,key,record))
複製代碼

這個接口，可讓使用者很是方便無需增長代碼的狀況下來實現指定每一個消息流綁定一個partition的結果。用戶也能夠經過本身實現一個partition的算法，來實現更精準的partition分配控制。具體實現能夠參考"kafka 指定partition生產，消費"中的介紹。