1.概述

最近和一些同學交流的時候反饋說，在面試Kafka時，被問到Kafka組件組成部分、API使用、Consumer和Producer原理及作用等問題都能詳細作答。但是，問到一個平時不注意的問題，就是Kafka的冪等性，被卡主了。那麼，今天筆者就為大家來剖析一下Kafka的冪等性原理及實現。

2.內容

2.1 Kafka為啥需要冪等性？

Producer在生產發送消息時，難免會重複發送消息。Producer進行retry時會產生重試機制，發生消息重複發送。而引入冪等性后，重複發送只會生成一條有效的消息。Kafka作為分佈式消息系統，它的使用場景常見與分佈式系統中，比如消息推送系統、業務平台系統（如物流平台、銀行結算平台等）。以銀行結算平台來說，業務方作為上游把數據上報到銀行結算平台，如果一份數據被計算、處理多次，那麼產生的影響會很嚴重。

2.2 影響Kafka冪等性的因素有哪些？

在使用Kafka時，需要確保Exactly-Once語義。分佈式系統中，一些不可控因素有很多，比如網絡、OOM、FullGC等。在Kafka Broker確認Ack時，出現網絡異常、FullGC、OOM等問題時導致Ack超時，Producer會進行重複發送。可能出現的情況如下：

2.3 Kafka的冪等性是如何實現的？

Kafka為了實現冪等性，它在底層設計架構中引入了ProducerID和SequenceNumber。那這兩個概念的用途是什麼呢？

• ProducerID：在每個新的Producer初始化時，會被分配一個唯一的ProducerID，這個ProducerID對客戶端使用者是不可見的。
• SequenceNumber：對於每個ProducerID，Producer發送數據的每個Topic和Partition都對應一個從0開始單調遞增的SequenceNumber值。

2.3.1 冪等性引入之前的問題？

Kafka在引入冪等性之前，Producer向Broker發送消息，然後Broker將消息追加到消息流中后給Producer返回Ack信號值。實現流程如下：

上圖的實現流程是一種理想狀態下的消息發送情況，但是實際情況中，會出現各種不確定的因素，比如在Producer在發送給Broker的時候出現網絡異常。比如以下這種異常情況的出現：

上圖這種情況，當Producer第一次發送消息給Broker時，Broker將消息(x2,y2)追加到了消息流中，但是在返回Ack信號給Producer時失敗了（比如網絡異常） 。此時，Producer端觸發重試機制，將消息(x2,y2)重新發送給Broker，Broker接收到消息后，再次將該消息追加到消息流中，然後成功返回Ack信號給Producer。這樣下來，消息流中就被重複追加了兩條相同的(x2,y2)的消息。

2.3.2 冪等性引入之後解決了什麼問題？

面對這樣的問題，Kafka引入了冪等性。那麼冪等性是如何解決這類重複發送消息的問題的呢？下面我們可以先來看看流程圖：

 同樣，這是一種理想狀態下的發送流程。實際情況下，會有很多不確定的因素，比如Broker在發送Ack信號給Producer時出現網絡異常，導致發送失敗。異常情況如下圖所示：

 當Producer發送消息(x2,y2)給Broker時，Broker接收到消息並將其追加到消息流中。此時，Broker返回Ack信號給Producer時，發生異常導致Producer接收Ack信號失敗。對於Producer來說，會觸發重試機制，將消息(x2,y2)再次發送，但是，由於引入了冪等性，在每條消息中附帶了PID（ProducerID）和SequenceNumber。相同的PID和SequenceNumber發送給Broker，而之前Broker緩存過之前發送的相同的消息，那麼在消息流中的消息就只有一條(x2,y2)，不會出現重複發送的情況。

2.3.3 ProducerID是如何生成的？

客戶端在生成Producer時，會實例化如下代碼：

// 實例化一個Producer對象
Producer<String, String> producer = new KafkaProducer<>(props);

在org.apache.kafka.clients.producer.internals.Sender類中，在run()中有一個maybeWaitForPid()方法，用來生成一個ProducerID，實現代碼如下：

 private void maybeWaitForPid() {
        if (transactionState == null)
            return;

        while (!transactionState.hasPid()) {
            try {
                Node node = awaitLeastLoadedNodeReady(requestTimeout);
                if (node != null) {
                    ClientResponse response = sendAndAwaitInitPidRequest(node);
                    if (response.hasResponse() && (response.responseBody() instanceof InitPidResponse)) {
                        InitPidResponse initPidResponse = (InitPidResponse) response.responseBody();
                        transactionState.setPidAndEpoch(initPidResponse.producerId(), initPidResponse.epoch());
                    } else {
                        log.error("Received an unexpected response type for an InitPidRequest from {}. " +
                                "We will back off and try again.", node);
                    }
                } else {
                    log.debug("Could not find an available broker to send InitPidRequest to. " +
                            "We will back off and try again.");
                }
            } catch (Exception e) {
                log.warn("Received an exception while trying to get a pid. Will back off and retry.", e);
            }
            log.trace("Retry InitPidRequest in {}ms.", retryBackoffMs);
            time.sleep(retryBackoffMs);
            metadata.requestUpdate();
        }
    }

3.事務

與冪等性有關的另外一個特性就是事務。Kafka中的事務與數據庫的事務類似，Kafka中的事務屬性是指一系列的Producer生產消息和消費消息提交Offsets的操作在一個事務中，即原子性操作。對應的結果是同時成功或者同時失敗。

這裏需要與數據庫中事務進行區別，操作數據庫中的事務指一系列的增刪查改，對Kafka來說，操作事務是指一系列的生產和消費等原子性操作。

3.1 Kafka引入事務的用途？

在事務屬性引入之前，先引入Producer的冪等性，它的作用為：

• Producer多次發送消息可以封裝成一個原子性操作，即同時成功，或者同時失敗；
• 消費者&生產者模式下，因為Consumer在Commit Offsets出現問題時，導致重複消費消息時，Producer重複生產消息。需要將這個模式下Consumer的Commit Offsets操作和Producer一系列生產消息的操作封裝成一個原子性操作。

產生的場景有：

比如，在Consumer中Commit Offsets時，當Consumer在消費完成時Commit的Offsets為100（假設最近一次Commit的Offsets為50），那麼執行觸發Balance時，其他Consumer就會重複消費消息（消費的Offsets介於50~100之間的消息）。

3.2 事務提供了哪些可使用的API？

Producer提供了五種事務方法，它們分別是：initTransactions()、beginTransaction()、sendOffsetsToTransaction()、commitTransaction()、abortTransaction()，代碼定義在org.apache.kafka.clients.producer.Producer<K,V>接口中，具體定義接口如下：

// 初始化事務，需要注意確保transation.id屬性被分配
void initTransactions();

// 開啟事務
void beginTransaction() throws ProducerFencedException;

// 為Consumer提供的在事務內Commit Offsets的操作
void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
                              String consumerGroupId) throws ProducerFencedException;

// 提交事務
void commitTransaction() throws ProducerFencedException;

// 放棄事務，類似於回滾事務的操作
void abortTransaction() throws ProducerFencedException;

3.3 事務的實際應用場景有哪些？

在Kafka事務中，一個原子性操作，根據操作類型可以分為3種情況。情況如下：

• 只有Producer生產消息，這種場景需要事務的介入；
• 消費消息和生產消息並存，比如Consumer&Producer模式，這種場景是一般Kafka項目中比較常見的模式，需要事務介入；
• 只有Consumer消費消息，這種操作在實際項目中意義不大，和手動Commit Offsets的結果一樣，而且這種場景不是事務的引入目的。

4.總結

Kafka的冪等性和事務是比較重要的特性，特別是在數據丟失和數據重複的問題上非常重要。Kafka引入冪等性，設計的原理也比較好理解。而事務與數據庫的事務特性類似，有數據庫使用的經驗對理解Kafka的事務也比較容易接受。

5.結束語

這篇博客就和大家分享到這裏，如果大家在研究學習的過程當中有什麼問題，可以加群進行討論或發送郵件給我，我會盡我所能為您解答，與君共勉！

另外，博主出書了《》和《》，喜歡的朋友或同學， 可以在公告欄那裡點擊購買鏈接購買博主的書進行學習，在此感謝大家的支持。關注下面公眾號，根據提示，可免費獲取書籍的教學視頻。 

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理【其他文章推薦】

※台北網頁設計公司這麼多，該如何挑選?? 網頁設計報價省錢懶人包"嚨底家"

※網頁設計公司推薦更多不同的設計風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

背景

• Read the fucking source code! –By 魯迅
• A picture is worth a thousand words. –By 高爾基

說明：

1. Kernel版本：4.14
2. ARM64處理器，Contex-A53，雙核
3. 使用工具：Source Insight 3.5， Visio

1. 概述

之前的文章分析的都是基於頁面的內存分配，而小塊內存的分配和管理是通過塊分配器來實現的。目前內核中，有三種方式來實現小塊內存分配：slab, slub, slob，最先有slab分配器，slub/slob分配器是改進版，slob分配器適用於小內存嵌入式設備，而slub分配器目前已逐漸成為主流塊分配器。接下來的文章，就是以slub分配器為目標，進一步深入。

先來一個初印象：

2. 數據結構

有四個關鍵的數據結構：

• struct kmem_cache：用於管理SLAB緩存，包括該緩存中對象的信息描述，per-CPU/Node管理slab頁面等；
  關鍵字段如下：

/*
 * Slab cache management.
 */
struct kmem_cache {
    struct kmem_cache_cpu __percpu *cpu_slab;       //每個CPU slab頁面
    /* Used for retriving partial slabs etc */
    unsigned long flags;
    unsigned long min_partial;
    int size;       /* The size of an object including meta data */
    int object_size;    /* The size of an object without meta data */
    int offset;     /* Free pointer offset. */
#ifdef CONFIG_SLUB_CPU_PARTIAL
    /* Number of per cpu partial objects to keep around */
    unsigned int cpu_partial;
#endif
    struct kmem_cache_order_objects oo;     //該結構體會描述申請頁面的order值，以及object的個數

    /* Allocation and freeing of slabs */
    struct kmem_cache_order_objects max;
    struct kmem_cache_order_objects min;
    gfp_t allocflags;   /* gfp flags to use on each alloc */
    int refcount;       /* Refcount for slab cache destroy */
    void (*ctor)(void *);           // 對象構造函數
    int inuse;      /* Offset to metadata */
    int align;      /* Alignment */
    int reserved;       /* Reserved bytes at the end of slabs */
    int red_left_pad;   /* Left redzone padding size */
    const char *name;   /* Name (only for display!) */
    struct list_head list;  /* List of slab caches */       //kmem_cache最終會鏈接在一個全局鏈表中
    struct kmem_cache_node *node[MAX_NUMNODES];     //Node管理slab頁面
};

• struct kmem_cache_cpu：用於管理每個CPU的slab頁面，可以使用無鎖訪問，提高緩存對象分配速度；

struct kmem_cache_cpu {
    void **freelist;    /* Pointer to next available object */                  //指向空閑對象的指針
    unsigned long tid;  /* Globally unique transaction id */                
    struct page *page;  /* The slab from which we are allocating */     //slab緩存頁面
#ifdef CONFIG_SLUB_CPU_PARTIAL
    struct page *partial;   /* Partially allocated frozen slabs */
#endif
#ifdef CONFIG_SLUB_STATS
    unsigned stat[NR_SLUB_STAT_ITEMS];
#endif
};

• struct kmem_cache_node：用於管理每個Node的slab頁面，由於每個Node的訪問速度不一致，slab頁面由Node來管理；

/*
 * The slab lists for all objects.
 */
struct kmem_cache_node {
    spinlock_t list_lock;

#ifdef CONFIG_SLUB
    unsigned long nr_partial;    //slab頁表數量
    struct list_head partial;       //slab頁面鏈表
#ifdef CONFIG_SLUB_DEBUG
    atomic_long_t nr_slabs;
    atomic_long_t total_objects;
    struct list_head full;
#endif
#endif
};

• struct page：用於描述slab頁面，struct page結構體中很多字段都是通過union聯合體進行復用的。
  struct page結構中，用於slub的成員如下：

struct page {
    union {
       ...
        void *s_mem;            /* slab first object */
       ...
    };
    
    /* Second double word */
    union {
       ...
        void *freelist;     /* sl[aou]b first free object */
       ...
    };
    
    union {
       ...
        struct {
            union {
              ...
                struct {            /* SLUB */
                    unsigned inuse:16;
                    unsigned objects:15;
                    unsigned frozen:1;
                };
                ...
            };
       ...
        };       
    };   
    
    /*
     * Third double word block
     */
    union {
       ...
        struct {        /* slub per cpu partial pages */
            struct page *next;  /* Next partial slab */
#ifdef CONFIG_64BIT
            int pages;  /* Nr of partial slabs left */
            int pobjects;   /* Approximate # of objects */
#else
            short int pages;
            short int pobjects;
#endif
        };

        struct rcu_head rcu_head;   /* Used by SLAB
                         * when destroying via RCU
                         */
    };
    ...
        struct kmem_cache *slab_cache;  /* SL[AU]B: Pointer to slab */    
    ...
}

圖來了：

3. 流程分析

針對Slub的使用，可以從三個維度來分析：

1. slub緩存創建
2. slub對象分配
3. slub對象釋放

下邊將進一步分析。

3.1 kmem_cache_create

在內核中通過kmem_cache_create接口來創建一個slab緩存。

先看一下這個接口的函數調用關係圖：

1. kmem_cache_create完成的功能比較簡單，就是創建一個用於管理slab緩存的kmem_cache結構，並對該結構體進行初始化，最終添加到全局鏈表中。kmem_cache結構體初始化，包括了上文中分析到的kmem_cache_cpu和kmem_cache_node兩個字段結構。

2. 在創建的過程中，當發現已有的slab緩存中，有存在對象大小相近，且具有兼容標誌的slab緩存，那就只需要進行merge操作並返回，而無需進一步創建新的slab緩存。

3. calculate_sizes函數會根據指定的force_order或根據對象大小去計算kmem_cache結構體中的size/min/oo等值，其中kmem_cache_order_objects結構體，是由頁面分配order值和對象數量兩者通過位域拼接起來的。

4. 在創建slab緩存的時候，有一個先雞后蛋的問題：kmem_cache結構體來管理一個slab緩存，而創建kmem_cache結構體又是從slab緩存中分配出來的對象，那麼這個問題是怎麼解決的呢？可以看一下kmem_cache_init函數，內核中定義了兩個靜態的全局變量kmem_cache和kmem_cache_node，在kmem_cache_init函數中完成了這兩個結構體的初始化之後，相當於就是創建了兩個slab緩存，一個用於分配kmem_cache結構體對象的緩存池，一個用於分配kmem_cache_node結構體對象的緩存池。由於kmem_cache_cpu結構體是通過__alloc_percpu來分配的，因此不需要創建一個相關的slab緩存。

3.2 kmem_cache_alloc

kmem_cache_alloc接口用於從slab緩存池中分配對象。

看一下大體的調用流程圖：

從上圖中可以看出，分配slab對象與Buddy System中分配頁面類似，存在快速路徑和慢速路徑兩種，所謂的快速路徑就是per-CPU緩存，可以無鎖訪問，因而效率更高。

整體的分配流程大體是這樣的：優先從per-CPU緩存中進行分配，如果per-CPU緩存中已經全部分配完畢，則從Node管理的slab頁面中遷移slab頁到per-CPU緩存中，再重新分配。當Node管理的slab頁面也不足的情況下，則從Buddy System中分配新的頁面，添加到per-CPU緩存中。

還是用圖來說明更清晰，分為以下幾步來分配：

1. fastpath
   快速路徑下，以原子的方式檢索per-CPU緩存的freelist列表中的第一個對象，如果freelist為空並且沒有要檢索的對象，則跳入慢速路徑操作，最後再返回到快速路徑中重試操作。

2. slowpath-1
   將per-CPU緩存中page指向的slab頁中的空閑對象遷移到freelist中，如果有空閑對象，則freeze該頁面，沒有空閑對象則跳轉到slowpath-2。

3. slowpath-2
   將per-CPU緩存中partial鏈表中的第一個slab頁遷移到page指針中，如果partial鏈表為空，則跳轉到slowpath-3。

4. slowpath-3
   將Node管理的partial鏈表中的slab頁遷移到per-CPU緩存中的page中，並重複第二個slab頁將其添加到per-CPU緩存中的partial鏈表中。如果遷移的slab中空閑對象超過了kmem_cache.cpu_partial的一半，則僅遷移slab頁，並且不再重複。
   如果每個Node的partial鏈表都為空，跳轉到slowpath-4。

5. slowpath-4
   從Buddy System中獲取頁面，並將其添加到per-CPU的page中。

3.2 kmem_cache_free

kmem_cache_free的操作，可以看成是kmem_cache_alloc的逆過程，因此也分為快速路徑和慢速路徑兩種方式，同時，慢速路徑中又分為了好幾種情況，可以參考kmem_cache_alloc的過程。

調用流程圖如下：

效果如下：

1. 快速路徑釋放
   快速路徑下，直接將對象返回到freelist中即可。

2. put_cpu_partial
   put_cpu_partial函數主要是將一個剛freeze的slab頁，放入到partial鏈表中。
   在put_cpu_partial函數中調用unfreeze_partials函數，這時候會將per-CPU管理的partial鏈表中的slab頁面添加到Node管理的partial鏈表的尾部。如果超出了Node的partial鏈表，溢出的slab頁面中沒有分配對象的slab頁面將會返回到夥伴系統。

3. add_partial
   添加slab頁到Node的partial鏈表中。

4. remove_partial
   從Node的partial鏈表移除slab頁。

具體釋放的流程走哪個分支，跟對象的使用情況，partial鏈表的個數nr_partial/min_partial等相關，細節就不再深入分析了。

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理【其他文章推薦】

※台北網頁設計公司這麼多，該如何挑選?? 網頁設計報價省錢懶人包"嚨底家"

※網頁設計公司推薦更多不同的設計風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整