鎖的機制

鎖和人很像，有的人樂觀，總會想到好的一方面，所以只要越努力，就會越幸運；有的人悲觀，總會想到不好的一方面，患得患失，所以經常會做不好事。我一直把前一個當作為我前進的動力和方向，快樂充實的過好每一天。

常用的鎖機制也有兩種：
1、樂觀鎖：假設不會發生并發沖突，每次不加鎖而去完成某項操作，只在提交操作時，檢查是否違反數據完整性。如果因為沖突失敗就繼續重試，直到成功為止。而樂觀鎖用到的機制就是CAS。

樂觀鎖大多是基于數據版本記錄機制實現。為數據增加一個版本標識，比如在基于數據庫表的版本解決方案中，一般是通過微數據庫表增加一個“version”字段來實現。讀取數據時，將此版本號一同讀出，之后更新時，對此版本號加一。此時，將提交數據的版本數據與數據庫表對應記錄的當前版本信息進行比對，如果提交的數據版本號大于數據庫表當前版本號，則予以更新，否則認為是過期數據。*樂觀鎖的缺點是不能解決臟讀的問題*。

注意：在實際生產環境里邊，如果并發量不大且不允許臟讀，可以使用悲觀鎖解決并發問題。但如果系統的并發非常大的話，悲觀鎖定會帶來非常大的性能問題，所以我們就要選擇樂觀鎖。

2、悲觀鎖：假定會發生并發沖突，屏蔽一切可能違反數據完整性的操作。悲觀鎖的實現，往往依靠底層提供的鎖機制。悲觀鎖會導致其他所有需要鎖的線程掛起，等待持有鎖的線程釋放鎖。如果所有線程都在等待其他線程釋放鎖，而不能主動釋放鎖資源，那么也會造成死鎖問題。

鎖的機制存在以下問題：
（1）在多線程競爭下，加鎖、釋放鎖會導致比較多的上下文切換和調度延時，引起性能問題。
（2）一個線程持有鎖會導致其他所有需要次所的線程掛起。
（3）如果一個優先級搞得線程等待一個優先級低的線程釋放鎖會導致優先級倒置，引起性能風險。

CAS操作

Compare And Set（或Compare And Swap），CAS是解決多線程并行情況下使用鎖造成性能損耗的一種機制，CAS操作包含三個操作數——內存位置（V）、預期原值（A）、新值（B）。如果內存位置的值與預期原值相同，那么處理器會自動將內存的值更新為新值。否則，處理器不做任何操作。無論哪種情況，處理器都會在CAS指令之前返回該位置的值。CAS有效地說明了“我認為位置V應該包含值A；如果包含該值，則將B放到這個位置；否則，不要更新該位置，只告訴我這個位置現在的值即可。”
現在幾乎所有的CPU指令都支持CAS的原子操作，X86下對應的是CMPXCHG匯編指令。有了這個操作，我們就可以用其來實現各種無鎖的數據結構。

這個操作可以用以下的例子來描述：
意思是，看一看內存*reg里的值是不是oldval，如果是的話，則對其賦值newval，并返回true，表示更新成功，如果返回false，則表示修改失敗。

bool compare_and_swap(int *reg,int oldval,int newval) {int reg_val = *reg;if(reg_val == oldval){*reg = newval;return true;}return false; }

CAS操作無鎖隊列的實現（參考）

Q:CAS的實現
A:gcc提供了兩個函數

bool __sync_bool_compare_and_swap (type *ptr, type oldval, type newval, ...)； type __sync_val_compare_and_swap (type *ptr, type oldval, type newval, ...)；

這兩個函數提供原子的比較和交換，如果*ptr == oldval,就將newval寫入*ptr,
第一個函數在相等并寫入的情況下返回true，這個函數比第二個好在，返回bool值可以知道有沒有更新成功。
第二個函數在返回操作之前的值。

第二個函數用c語言描述：

type __sync_val_compare_and_swap (type *ptr, type oldval, type newval, ...) {type cur = *ptr;if (cur == oldval){*ptr = newval;}return cur;// 返回操作之前的值 }

type只能是1,2,4或8字節長度的int類型，否則會發生下面的錯誤

Q: 操作系統級別是如何實現的
A: X86中有一個CMPXCHG的匯編指令

Q: CAS指令有什么缺點
A:
1.存在ABA問題因為CAS需要在操作值的時候檢查下值有沒有發生變化，如果沒有發生變化則更新，但是如果一個值原來是A，變成了B，又變成了A，那么使用CAS進行檢查時會發現它的值沒有發生變化，但是實際上卻變化了。ABA問題的解決思路就是使用版本號。在變量前面追加上版本號，每次變量更新的時候把版本號加一，那么A－B－A 就會變成1A-2B-3A。
2.循環時間長開銷大自旋CAS如果長時間不成功，會給CPU帶來非常大的執行開銷。
3.只能保證一個共享變量的原子操作對一個共享變量執行操作時，我們可以使用循環CAS的方式來保證原子操作，但是對多個共享變量操作時，循環CAS就無法保證操作的原子性，這個時候就可以用鎖，或者有一個取巧的辦法，就是把多個共享變量合并成一個共享變量來操作。比如有兩個共享變量i＝2,j=a，合并一下ij=2a，然后用CAS來操作ij。

gcc從4.1.2提供了__sync_*系列的built-in函數，用于提供加減和邏輯運算的原子操作。

其聲明如下：

原子操作的后置加加type __sync_fetch_and_add (type *ptr, type value, …)
原子操作的前置加加type __sync_add_and_fetch (type *ptr, type value, …)
其他類比
type __sync_fetch_and_sub (type *ptr, type value, …)
type __sync_fetch_and_or (type *ptr, type value, …)
type __sync_fetch_and_and (type *ptr, type value, …)
type __sync_fetch_and_xor (type *ptr, type value, …)
type __sync_fetch_and_nand (type *ptr, type value, …)

type __sync_sub_and_fetch (type *ptr, type value, …)
type __sync_or_and_fetch (type *ptr, type value, …)
type __sync_and_and_fetch (type *ptr, type value, …)
type __sync_xor_and_fetch (type *ptr, type value, …)
type __sync_nand_and_fetch (type *ptr, type value, …)
這兩組函數的區別在于第一組返回更新前的值，第二組返回更新后的值。

關于CAS函數，參考: http://blog.csdn.net/youfuchen/article/details/23179799

在多線程環境下有以下情景：

比如對同一鏈隊列進行入隊操作時一個線程正在將新的隊列節點 掛載到 隊尾節點的next上，可是還沒來的及更新隊尾節點 但同一時刻另一個線程也在進行入隊操作將新的隊列節點也掛在到了沒更新的隊尾節點那么先掛載的節點就丟失了。

為了解決多線程環境下的這些問題，我們第一時間肯定想到了加上互斥鎖控制同一時刻只能有一個線程可以對隊列進行寫操作，但是加鎖的操作太消耗系統資源了很繁重 。

因為對臨界區的操作只有一步 就是對隊列的尾節點進行更新，只要讓這一步進行的是原子操作就可以了，所以使用到了CAS操作。

為了有一個對比 寫了一份thread_queue.c是用鎖對臨界區進行控制訪問的
另一份是lock_free_queue.c是用CAS確保對臨界區的操作是原子操作

“queue.h”#ifndef QUEUE_H_ #define QUEUE_H_#include <stdio.h> #include <stdlib.h>/* 普通的 鏈式隊列 */ typedef struct QNode {int data;struct QNode *next; }QNode, *QueuePtr;typedef struct LinkQueue {QueuePtr front;QueuePtr rear; }LinkQueue;void init_Queue(LinkQueue *q);//初始化隊列 void push_Queue(LinkQueue *q, int e);//隊尾入隊 int pop_Queue(LinkQueue *q, int *e);//隊頭出隊 int is_Empty(LinkQueue *q); void show(LinkQueue *q);#endif /* QUEUE_H_ */“queue.c”#include "queue.h"/* 初始化 為隊列構建一個頭結點 讓front和rear都指向這個頭結點 */ void init_Queue(LinkQueue *q) {q->front = q->rear = (QNode *)malloc(sizeof(QNode));q->front->next = NULL; }/* 普通的入隊操作 */ void push_Queue(LinkQueue *q, int e) {QueuePtr newNode = (QueuePtr)malloc(sizeof(QNode));newNode->data = e;newNode->next = NULL;q->rear->next = newNode;q->rear = newNode; }/* cas的入隊操作 和普通的入隊操作一樣 新建節點后 要將新節點掛在隊尾時需要進行cas操作 因為官方文檔：The definition given in the Intel documentation allows only for the use of the types int, long, long long as well as their unsigned counterparts 只能用 int, long, long long 所以要把指針類型 QueuePtr 變成 long 用long的另一個原因就是：屏蔽32位和64位的差異 long在32位是4字節 64位是8字節 */ void cas_push(LinkQueue *q, int e) {QueuePtr newNode = (QueuePtr)malloc(sizeof(QNode));newNode->data = e;newNode->next = NULL;QueuePtr tmp;do{tmp = q->rear;}while (!__sync_bool_compare_and_swap((long *)(&(tmp->next)), NULL, (long)newNode));q->rear = newNode; }/* 以前的判空是 q->front == q->rear 但是這樣子會增加出隊的操作 當出的是最后一個元素時, q->rear需要指向 q->front 我把這一步省了 暫時沒有發現有什么副作用 所以我改成了 q->front->next == NULL */ int is_Empty(LinkQueue *q) {if (q->front->next == NULL){return(1);}return(0); }/* 普通的出隊操作 如果隊空 返回0 也就是false e作為接受元素的緩沖 */ int pop_Queue(LinkQueue *q, int *e) {if (is_Empty(q)){return(0);}QueuePtr tmp;tmp = q->front->next;q->front->next = tmp->next;*e = tmp->data;free(tmp);return(1); }/* cas的出隊操作 每一次都要判斷這個隊列是不是空 然后執行cas的出隊操作： （1）tmp = q->rear 把舊的隊頭存起來 （2）執行原子操作：看 舊的隊頭 是否等于 現在的隊頭 tmp == *(&(q->front)) 如果相等執行 *(&(q->front)) = tmp->next 返回true 否則，即執行這一步原子操作的時候，別的線程修改了隊列，導致隊尾指向改變了，返回false ,while(!false)回到第一步重新執行 */ int cas_pop(LinkQueue *q, int *e) {QueuePtr tmp;do {if (is_Empty(q)){return(0);}//printf("cas_pop...\n");tmp = q->front->next;} while (!__sync_bool_compare_and_swap((long *)(&(q->front->next)), (long)tmp, (long)tmp->next));*e = tmp->data;free(tmp);return(1); }/* 遍歷隊列 打印里面的元素 為了求證隊列里面的元素 */ void show(LinkQueue *q) {printf("void show(LinkQueue *q)\n");QueuePtr tmp = q->front->next;while (tmp){printf("%d ", tmp->data);tmp = tmp->next;}printf("\n"); }“thread_queue.c”#include "queue.h" #include <pthread.h> #include <unistd.h> #include <semaphore.h> #include <assert.h>#define THREAD_NUMBER 4//開啟的線程數，電腦是4核，所以用4//sem_t queue_sem;//信號量 pthread_mutex_t mutex;//互斥鎖void *thread_push(void *arg); void *thread_pop(void *arg);int main() {LinkQueue que;init_Queue(&que);/*初始化二進制信號量 初始值為1 代表每一次只有1個線程可以訪問 本來更加應該用互斥量 比較貼合情景 但是不太熟 就用了信號量*///int res = sem_init(&queue_sem, 0, 1);//assert(res != -1);int i;pthread_t threadArr[THREAD_NUMBER];for (i = 0; i < THREAD_NUMBER; ++i){pthread_create(&threadArr[i], NULL, thread_push, (void *)&que);}for (i = 0; i < THREAD_NUMBER; ++i){pthread_join(threadArr[i], NULL);}show(&que);for (i = 0; i < THREAD_NUMBER; ++i){pthread_create(&threadArr[i], NULL, thread_pop, (void *)&que);}for (i = 0; i < THREAD_NUMBER; ++i){pthread_join(threadArr[i], NULL);}//sem_destroy(&queue_sem);exit(EXIT_SUCCESS); }void *thread_push(void *arg) {printf("start push\n");LinkQueue * quePtr = (LinkQueue *)arg;int i;for (i = 0; i < 20; ++i){//sem_wait(&queue_sem);pthread_mutex_lock(&mutex);push_Queue(quePtr, i);pthread_mutex_unlock(&mutex);//sem_post(&queue_sem);}printf("finish push\n");pthread_exit(NULL); }void *thread_pop(void *arg) {printf("start pop\n");LinkQueue * quePtr = (LinkQueue *)arg;int tmp;int res;while (1){//sem_wait(&queue_sem);pthread_mutex_lock(&mutex);res = pop_Queue(quePtr, &tmp);pthread_mutex_unlock(&mutex);//sem_post(&queue_sem);if (!res){break;}printf("%d ", tmp);}printf("finish pop\n");pthread_exit(NULL); }“lock_free_queue.c”#include "queue.h" #include <pthread.h> #include <unistd.h> #include <assert.h>#define THREAD_NUMBER 4//開啟的線程數，電腦是4核，所以用4void *thread_push(void *arg); void *thread_pop(void *arg);/* 初始化空隊列為了模擬線程對資源的搶占 開啟4個線程 每個線程push 20個元素 0~19 等待4個線程結束 打印隊列元素 驗證push 開啟四個線程 每個線程都對隊列進行 pop操作 */ int main() {LinkQueue que;init_Queue(&que);int i;/*創造四個新線程 每個線程都執行 thread_push(&que)*/pthread_t threadArr[THREAD_NUMBER];for (i = 0; i < THREAD_NUMBER; ++i){pthread_create(&threadArr[i], NULL, thread_push, (void *)&que);}/*等待四個線程都執行完要不然主線程一下子就跑完了 程序就結束了還有就是 為了show函數 可以驗證元素是不是都push進去了*/for (i = 0; i < THREAD_NUMBER; ++i){pthread_join(threadArr[i], NULL);}show(&que);/*創造四個新線程 每個線程都執行 thread_pop(&que)*/for (i = 0; i < THREAD_NUMBER; ++i){pthread_create(&threadArr[i], NULL, thread_pop, (void *)&que);}for (i = 0; i < THREAD_NUMBER; ++i){pthread_join(threadArr[i], NULL);}exit(EXIT_SUCCESS); }void *thread_push(void *arg) {printf("start push\n");LinkQueue * quePtr = (LinkQueue *)arg;int i;for (i = 0; i < 20; ++i){cas_push(quePtr, i);}printf("finish push\n");pthread_exit(NULL); }void *thread_pop(void *arg) {printf("start pop\n");LinkQueue * quePtr = (LinkQueue *)arg;int tmp;int res;while (1){res = cas_pop(quePtr, &tmp);if (!res){break;}printf("%d ", tmp);//sleep(1);}printf("finish pop\n");pthread_exit(NULL); }

總結

以上是生活随笔為你收集整理的锁、CAS操作和无锁队列的实现的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。