利用C++ 設計緩存隊列實現高效傳輸相機數據(SampleBuffer)

利用C++ 設計緩存隊列實現高效傳輸相機數據

需求：

在做例如直播功能，有時我們可能要對相機捕獲的圖像數據做一些額外操作(Crop, Scale, 美顏等)但由于某些操作算法本身很耗時明肮，以fps為30為例能真，如果某一幀處理較慢將可能會掉幀流炕，所以設計一個緩沖隊列先將捕獲到的相機數據放入空閑隊列中赁遗，隨后程序中如果需要使用到相機數據則從工作隊列中取出需要的數據曼追。

適用情況

在相機回調中對每一幀圖像進行耗時操作(Crop, Scale...)
提升處理圖像的效率
高效處理其他大數據量工作

注意：本例通過設計使用C++ 隊列來實現相機SampleBuffer的緩存工作顽决，需要使用Objective-C 與 C++混編短条。

GitHub地址(附代碼) : C++緩存隊列

簡書地址 : C++緩存隊列

博客地址 : C++緩存隊列

掘金地址 : C++緩存隊列

總體流程：

設置始終橫屏，初始化相機參數設置代理
在捕捉相機數據的回調中將samplebuffer放入空閑隊列
開啟一條線程每隔10ms從工作隊列中取出samplebuffer可在此對數據處理才菠，處理完后將結點放回空閑隊列

隊列實現及解析

1.原理

初始化固定數量的結點裝入空閑隊列茸时，當相機回調產生數據后，從空閑隊列頭部取出一個結點將產生的每一幀圖像buffer裝入赋访，然后入隊到工作隊列的尾部可都，處理buffer的線程從工作隊列的頭部取出一個結點中的Buffer進行處理，處理完成后會將裝有次buffer的結點中data置空并重新放入空閑隊列的頭部以供下次使用蚓耽。

原理.png

解析

我們將空閑隊列設計為頭進頭出渠牲，影響不大，因為我們每次只需要從空閑隊列中取出一個空結點以供我們裝入相機數據步悠，所以沒必要按照尾進頭出的方式保證結點的順序签杈。
我們將工作隊列設計為尾進頭出，因為我們要確保從相機中捕獲的數據是連續(xù)的鼎兽，以便后期我們播放出來的畫面也是連續(xù)的答姥，所以工作隊列必須保證尾進頭出。
這樣做我們相當于實現了用空閑隊列當做緩沖隊列谚咬，在正常情況下(fps=30,即每秒產生30幀數據鹦付，大約每33ms產生一幀數據)，如果在33ms內對數據進行的操作可以正常完成择卦，則工作隊列會保持始終為0或1敲长，但是如果長期工作或遇到某一幀數據處理較慢的情況(即處理時間大于33ms)則工作隊列的長度會增加，而正因為我們使用了這樣的隊列會保護那一幀處理慢的數據在仍然能夠正常處理完秉继。

注意：這種情景僅用于短時間內僅有幾幀數據處理較慢潘明，如果比如1s內有20幾幀數據都處理很慢則可能導致工作隊列太長，則提現不出此隊列的優(yōu)勢秕噪。

2.結構

結點

typedef struct XDXCustomQueueNode {
    void    *data;
    size_t  size;  // data size
    long    index;
    struct  XDXCustomQueueNode *next;
} XDXCustomQueueNode;

結點中使用void *類型的data存放我們需要的sampleBuffer,使用index記錄當前裝入結點的sampleBuffer的索引钳降，以便我們在取出結點時比較是否是按照順序取出，結點中還裝著同類型下一個結點的元素腌巾。

隊列類型

typedef struct XDXCustomQueue {
    int size;
    XDXCustomQueueType type;
    XDXCustomQueueNode *front;
    XDXCustomQueueNode *rear;
} XDXCustomQueue;

隊列中即為我們裝載的結點數量遂填，因為我們采用的是預先分配固定內存铲觉，所以工作隊列與空閑隊列的和始終不變(因為結點中的元素不在工作隊列就在空閑隊列)

類的設計

class XDXCustomQueueProcess {
    
private:
    pthread_mutex_t free_queue_mutex;
    pthread_mutex_t work_queue_mutex;
    
public:
    XDXCustomQueue *m_free_queue;
    XDXCustomQueue *m_work_queue;
    
    XDXCustomQueueProcess();
    ~XDXCustomQueueProcess();
    
    // Queue Operation
    void InitQueue(XDXCustomQueue *queue,
                   XDXCustomQueueType type);
    void EnQueue(XDXCustomQueue *queue,
                 XDXCustomQueueNode *node);
    XDXCustomQueueNode *DeQueue(XDXCustomQueue *queue);
    void ClearXDXCustomQueue(XDXCustomQueue *queue);
    void FreeNode(XDXCustomQueueNode* node);
    void ResetFreeQueue(XDXCustomQueue *workQueue, XDXCustomQueue *FreeQueue);
};

因為涉及到異步操作，所以需要對結點的操作加鎖吓坚，使用時需要先初始化隊列撵幽，然后定義了入隊，出隊礁击，清除隊列中元素盐杂，釋放結點，重置空閑隊列等操作哆窿。

3.實現

初始化隊列

const int XDXCustomQueueSize = 3;
XDXCustomQueueProcess::XDXCustomQueueProcess() {
    m_free_queue = (XDXCustomQueue *)malloc(sizeof(struct XDXCustomQueue));
    m_work_queue = (XDXCustomQueue *)malloc(sizeof(struct XDXCustomQueue));
    
    InitQueue(m_free_queue, XDXCustomFreeQueue);
    InitQueue(m_work_queue, XDXCustomWorkQueue);
    
    for (int i = 0; i < XDXCustomQueueSize; i++) {
        XDXCustomQueueNode *node = (XDXCustomQueueNode *)malloc(sizeof(struct XDXCustomQueueNode));
        node->data = NULL;
        node->size = 0;
        node->index= 0;
        this->EnQueue(m_free_queue, node);
    }
    
    pthread_mutex_init(&free_queue_mutex, NULL);
    pthread_mutex_init(&work_queue_mutex, NULL);
    
    NSLog(@"XDXCustomQueueProcess Init finish !");
}

假設空閑隊列結點總數為3.首先為工作隊列與空閑隊列分配內存链烈，其次對其分別進行初始化操作，具體過程可參考Demo,然后根據結點總數來為每個結點初始化分配內存挚躯，并將分配好內存的結點入隊到空閑隊列中强衡。

注意：結點的重用，我們僅僅初始化幾個固定數量的結點码荔，因為處理數據量較大漩勤，沒有必要讓程序始終做malloc與free，為了優(yōu)化我們這里的隊列相當于一個靜態(tài)鏈表缩搅，即結點的復用越败，因為當結點在工作隊列中使用完成后會將其中的數據置空并重新入隊到空閑隊列中，所以結點的總數始終保持不變硼瓣。

入隊Enqueue

void XDXCustomQueueProcess::EnQueue(XDXCustomQueue *queue, XDXCustomQueueNode *node) {
    if (queue == NULL) {
        NSLog(@"XDXCustomQueueProcess Enqueue : current queue is NULL");
        return;
    }
    
    if (node==NULL) {
        NSLog(@"XDXCustomQueueProcess Enqueue : current node is NULL");
        return;
    }
    
    node->next = NULL;
    
    if (XDXCustomFreeQueue == queue->type) {
        pthread_mutex_lock(&free_queue_mutex);
        
        if (queue->front == NULL) {
            queue->front = node;
            queue->rear  = node;
        }else {
            /*
             // tail in,head out
             freeQueue->rear->next = node;
             freeQueue->rear = node;
             */
            
            // head in,head out
            node->next = queue->front;
            queue->front = node;
        }
        queue->size += 1;
        NSLog(@"XDXCustomQueueProcess Enqueue : free queue size=%d",queue->size);
        pthread_mutex_unlock(&free_queue_mutex);
    }
    
    if (XDXCustomWorkQueue == queue->type) {
        pthread_mutex_lock(&work_queue_mutex);
        //TODO
        static long nodeIndex = 0;
        node->index=(++nodeIndex);
        if (queue->front == NULL) {
            queue->front = node;
            queue->rear  = node;
        }else {
            queue->rear->next   = node;
            queue->rear         = node;
        }
        queue->size += 1;
        NSLog(@"XDXCustomQueueProcess Enqueue : work queue size=%d",queue->size);
        pthread_mutex_unlock(&work_queue_mutex);
    }
}

如上所述究飞，入隊操作如果是空閑隊列，則使用頭進的方式巨双，即始終讓入隊的結點在隊列的頭部噪猾，具體代碼實現即讓當前結點的next指向空閑隊列的頭結點霉祸，然后將當前結點變?yōu)榭臻e隊列的頭結點筑累；如果入隊操作是工作隊列，則使用尾進的方式丝蹭，并對結點的index賦值慢宗，以便我們在取出結點時可以打印Index是否連續(xù)，如果連續(xù)則說明入隊時始終保持順序入隊奔穿。

這里使用了簡單的數據結構中的知識镜沽，如有不懂可上網進行簡單查閱

出隊

XDXCustomQueueNode* XDXCustomQueueProcess::DeQueue(XDXCustomQueue *queue) {
    if (queue == NULL) {
        NSLog(@"XDXCustomQueueProcess DeQueue : current queue is NULL");
        return NULL;
    }
    
    const char *type = queue->type == XDXCustomWorkQueue ? "work queue" : "free queue";
    pthread_mutex_t *queue_mutex = ((queue->type == XDXCustomWorkQueue) ? &work_queue_mutex : &free_queue_mutex);
    XDXCustomQueueNode *element = NULL;
    
    pthread_mutex_lock(queue_mutex);
    element = queue->front;
    if(element == NULL) {
        pthread_mutex_unlock(queue_mutex);
        NSLog(@"XDXCustomQueueProcess DeQueue : The node is NULL");
        return NULL;
    }
    
    queue->front = queue->front->next;
    queue->size -= 1;
    pthread_mutex_unlock(queue_mutex);
    
    NSLog(@"XDXCustomQueueProcess DeQueue : %s size=%d",type,queue->size);
    return element;
}

出隊操作無論空閑隊列還是工作隊列都是從頭出，即取出當前隊列頭結點中的數據贱田。

注意：該結點為空與該結點中的數據為空不可混為一談缅茉，如果該結點為空則說明沒有從隊列中取出結點，即空結點沒有內存地址男摧，而結點中的數據則為node->data,在本Demo中為相機產生的每一幀sampleBuffer數據蔬墩。

重置空閑隊列數據

void XDXCustomQueueProcess::ResetFreeQueue(XDXCustomQueue *workQueue, XDXCustomQueue *freeQueue) {
    if (workQueue == NULL) {
        NSLog(@"XDXCustomQueueProcess ResetFreeQueue : The WorkQueue is NULL");
        return;
    }
    
    if (freeQueue == NULL) {
        NSLog(@"XDXCustomQueueProcess ResetFreeQueue : The FreeQueue is NULL");
        return;
    }
    
    int workQueueSize = workQueue->size;
    if (workQueueSize > 0) {
        for (int i = 0; i < workQueueSize; i++) {
            XDXCustomQueueNode *node = DeQueue(workQueue);
            CFRelease(node->data);
            node->data = NULL;
            EnQueue(freeQueue, node);
        }
    }
    NSLog(@"XDXCustomQueueProcess ResetFreeQueue : The work queue size is %d, free queue size is %d",workQueue->size, freeQueue->size);
}

當我們將執(zhí)行一些中斷操作译打，例如從本View跳轉到其他View，或進入后臺等操作拇颅，我們需要將工作隊列中的結點均置空然后重新放回空閑隊列奏司，這樣可以保證我們最初申請的結點還均有效可用，保證結點不會丟失樟插。

流程

1.初始化相機相關參數

常規(guī)流程韵洋，Demo中有實現，在此不復述

2.將samplebuffer放入空閑隊列

設置相機代理后黄锤，在 - (void)captureOutput:(AVCaptureOutput *)captureOutput didOutputSampleBuffer:(CMSampleBufferRef)sampleBuffer fromConnection:(AVCaptureConnection *)connection 方法中將samplebuffer裝入空閑隊列

- (void)addBufferToWorkQueueWithSampleBuffer:(CMSampleBufferRef)sampleBuffer {
    XDXCustomQueueNode *node = _captureBufferQueue->DeQueue(_captureBufferQueue->m_free_queue);
    if (node == NULL) {
        NSLog(@"XDXCustomQueueProcess addBufferToWorkQueueWithSampleBuffer : Data in , the node is NULL !");
        return;
    }
    CFRetain(sampleBuffer);
    node->data = sampleBuffer;
    _captureBufferQueue->EnQueue(_captureBufferQueue->m_work_queue, node);

    NSLog(@"XDXCustomQueueProcess addBufferToWorkQueueWithSampleBuffer : Data in ,  work size = %d, free size = %d !",_captureBufferQueue->m_work_queue->size, _captureBufferQueue->m_free_queue->size);
}

注意：因為相機回調中捕捉的sampleBuffer是有生命周期的所以需要手動CFRetain一下使我們隊列中的結點持有它搪缨。

3.開啟一條線程處理隊列中的Buffer

使用pthread創(chuàng)建一條線程，每隔10ms取一次數據,我們可以在此對取到的數據進行我們想要的操作猜扮，操作完成后再將清空釋放sampleBuffer再將其裝入空閑隊列供我們循環(huán)使用勉吻。

- (void)handleCacheThread {
    while (true) {
        // 從隊列取出在相機回調中放入隊列的線程
        XDXCustomQueueNode *node = _captureBufferQueue->DeQueue(_captureBufferQueue->m_work_queue);
        if (node == NULL) {
            NSLog(@"Crop handleCropThread : Data node is NULL");
            usleep(10*1000);
            continue;
        }
        
        CMSampleBufferRef sampleBuffer     = (CMSampleBufferRef)node->data;
        // 打印結點的index，如果連續(xù)則說明在相機回調中放入的samplebuffer是連續(xù)的
        NSLog(@"Test index : %ld",node->index);
        
        /* 可在此處理從隊列中拿到的Buffer旅赢，用完后記得釋放內存并將結點重新放回空閑隊列
         * ........
         */
        
        CFRelease(sampleBuffer);
        node->data = NULL;
        _captureBufferQueue->EnQueue(_captureBufferQueue->m_free_queue, node);
    }
}