一般情況下，iOS開發(fā)者只要會(huì)使用GCD、@synchronized、NSLock等幾個(gè)簡(jiǎn)單的API，就可以應(yīng)對(duì)大部分多線程開發(fā)了，不過(guò)這樣是否真正做到了多線程安全，又是否真正充分利用了多線程的效率優(yōu)勢(shì)呢?看看以下幾個(gè)容易被忽略的細(xì)節(jié)。

讀者寫者問(wèn)題(Readers-writers problem)

先看下讀者寫者問(wèn)題的描述：

有讀者和寫者兩組并發(fā)線程，共享同一數(shù)據(jù)，當(dāng)兩個(gè)或以上的讀線程同時(shí)訪問(wèn)共享數(shù)據(jù)時(shí)不會(huì)產(chǎn)生副作用，但若某個(gè)寫線程和其他線程(讀線程或?qū)懢€程)同時(shí)訪問(wèn)共享數(shù)據(jù)時(shí)則可能導(dǎo)致數(shù)據(jù)不一致的錯(cuò)誤。因此要求：

• 允許多個(gè)讀者可以同時(shí)對(duì)共享數(shù)據(jù)執(zhí)行讀操作;
• 只允許一個(gè)寫者寫共享數(shù)據(jù);
• 任一寫者在完成寫操作之前不允許其他讀者或?qū)懻吖ぷ?
• 寫者執(zhí)行寫操作前，應(yīng)讓已有的讀者和寫者全部退出。

從以上描述可以得知，所謂“讀者寫者問(wèn)題”是指保證一個(gè)寫線程必須與其他線程互斥地訪問(wèn)共享對(duì)象的同步問(wèn)題，允許并發(fā)讀操作，但是寫操作必須和其他讀寫操作是互斥的。

大部分客戶端App做的事情無(wú)非就是從網(wǎng)絡(luò)拉取***數(shù)據(jù)、加工數(shù)據(jù)、展現(xiàn)列表，這個(gè)過(guò)程中既有拿到***數(shù)據(jù)后寫入本地的操作，也有上層業(yè)務(wù)對(duì)本地?cái)?shù)據(jù)的讀取操作，因此會(huì)牽涉大量的多線程讀寫操作，很顯然，這些基本都屬于讀者寫者問(wèn)題的范疇[1]。

然而筆者注意到，在遇到多線程讀寫問(wèn)題時(shí)，多數(shù)iOS開發(fā)者都會(huì)立即想到加鎖，或者干脆避免使用多線程，但卻少有人會(huì)嘗試用讀者寫者問(wèn)題的思路去進(jìn)一步提升效率。

以下是實(shí)現(xiàn)一個(gè)簡(jiǎn)單cache的示例代碼：

//實(shí)現(xiàn)一個(gè)簡(jiǎn)單的cache 
- (void)setCache:(id)cacheObject forKey:(NSString *)key { 
    if (key.length == 0) { 
        return; 
    } 
    [_cacheLock lock]; 
    self.cacheDic[key] = cacheObject; 
    ... 
    [_cacheLock unlock]; 
} 
 
- (id)cacheForKey:(NSString *key) { 
    if (key.length == 0) { 
        return nil; 
    } 
    [_cacheLock lock]; 
    id cacheObject = self.cacheDic[key]; 
    ... 
    [_cacheLock unlock]; 
    return cacheObject; 
} 

上述代碼用互斥鎖來(lái)實(shí)現(xiàn)多線程讀寫，做到了數(shù)據(jù)的安全讀寫，但是效率卻并不是***的，因?yàn)檫@種情況下，雖然寫操作和其他操作之間是互斥的，但同時(shí)讀操作之間卻也是互斥的，這會(huì)浪費(fèi)cpu資源，如何改良呢?不難發(fā)現(xiàn)，這其實(shí)是個(gè)典型的讀者寫者問(wèn)題。先看下解決讀者寫者問(wèn)題的偽代碼：

semaphore ReaderWriterMutex = 1;    //實(shí)現(xiàn)讀寫互斥 
int Rcount = 0;             //讀者數(shù)量 
semaphore CountMutex = 1;   //讀者修改計(jì)數(shù)互斥 
 
writer(){ 
    while(true){ 
        P(ReaderWriterMutex); 
        write; 
        V(ReaderWriterMutex);    
    } 
     
} 
 
reader(){ 
    while(true){ 
        P(CountMutex); 
        if(Rcount == 0)     //當(dāng)***個(gè)讀者進(jìn)來(lái)時(shí)，阻塞寫者 
            P(ReaderWriterMutex); 
        ++Rcount; 
        V(CountMutex); 
 
        read; 
 
        P(CountMutex); 
        --Rcount; 
        if(Rcount == 0) 
            V(ReaderWriterMutex);   //當(dāng)***一個(gè)讀者離開后，釋放寫者 
        V(CountMutex); 
    } 
} 

在iOS中，上述代碼中的PV原語(yǔ)可以替換成GCD中的信號(hào)量API，dispatch_semaphore_t來(lái)實(shí)現(xiàn)，但是需要額外維護(hù)一個(gè)readerCount以及實(shí)現(xiàn)readerCount互斥訪問(wèn)的信號(hào)量，手動(dòng)實(shí)現(xiàn)比較麻煩，封裝成統(tǒng)一接口有一定難度。不過(guò)好在iOS開發(fā)中可以找到現(xiàn)成的讀者寫者鎖：

pthread_rwlock_t

這是一個(gè)古老的C語(yǔ)言層面的函數(shù)，用法如下：

// Initialization of lock, pthread_rwlock_t is a value type and must be declared as var in order to refer it later. Make sure not to copy it. 
var lock = pthread_rwlock_t() 
pthread_rwlock_init(&lock, nil) 
 
// Protecting read section: 
pthread_rwlock_rdlock(&lock) 
// Read shared resource 
pthread_rwlock_unlock(&lock) 
 
// Protecting write section: 
pthread_rwlock_wrlock(&lock) 
// Write shared resource 
pthread_rwlock_unlock(&lock) 
 
// Clean up 
pthread_rwlock_destroy(&lock) 

接口簡(jiǎn)潔但是卻不友好，需要注意pthread_rwlock_t是值類型，用=賦值會(huì)直接拷貝，不小心就會(huì)浪費(fèi)內(nèi)存，另外用完后還需要記得銷毀，容易出錯(cuò)，有沒(méi)有更高級(jí)更易用的API呢?

GCD barrier

dispatch_barrier_async / dispatch_barrier_sync并不是專門用來(lái)解決讀者寫者問(wèn)題的，barrier主要用于以下場(chǎng)景：當(dāng)執(zhí)行某一任務(wù)A時(shí)，需要該隊(duì)列上之前添加的所有操作都執(zhí)行完，而之后添加進(jìn)來(lái)的任務(wù)，需要等待任務(wù)A執(zhí)行完畢才可以執(zhí)行，從而達(dá)到將任務(wù)A隔離的目的，具體過(guò)程如下圖所示：

如果將barrier任務(wù)之前和之后的并發(fā)任務(wù)換為讀操作，barrier任務(wù)本身?yè)Q為寫操作，就可以將dispatch_barrier_async / dispatch_barrier_sync當(dāng)做讀者寫者鎖來(lái)使用了，下面把文初的使用普通鎖實(shí)現(xiàn)的cache代碼，用dispatch_barrier_async重寫，做下對(duì)比：

//實(shí)現(xiàn)一個(gè)簡(jiǎn)單的cache（使用普通鎖） 
- (void)setCache:(id)cacheObject forKey:(NSString *)key { 
    if (key.length == 0) { 
        return; 
    } 
    [_cacheLock lock]; 
    self.cacheDic[key] = cacheObject; 
    ... 
    [_cacheLock unlock]; 
} 
 
- (id)cacheForKey:(NSString *key) { 
    if (key.length == 0) { 
        return nil; 
    } 
    [_cacheLock lock]; 
    id cacheObject = self.cacheDic[key]; 
    ... 
    [_cacheLock unlock]; 
    return cacheObject; 
} 

//實(shí)現(xiàn)一個(gè)簡(jiǎn)單的cache（使用讀者寫者鎖） 
static dispatch_queue_t queue = dispatch_queue_create("com.gfzq.testQueue", DISPATCH_QUEUE_CONCURRENT); 
 
- (void)setCache:(id)cacheObject forKey:(NSString *)key { 
    if (key.length == 0) { 
        return; 
    } 
    dispatch_barrier_async(queue, ^{ 
        self.cacheDic[key] = cacheObject; 
        ... 
    }); 
} 
 
- (id)cacheForKey:(NSString *key) { 
    if (key.length == 0) { 
        return nil; 
    } 
    __block id cacheObject = nil; 
    dispatch_sync(queue, ^{ 
        cacheObject = self.cacheDic[key]; 
        ... 
    }); 
    return cacheObject; 
} 

這樣實(shí)現(xiàn)的cache就可以并發(fā)執(zhí)行讀操作，同時(shí)又有效地隔離了寫操作，兼顧了安全和效率。

對(duì)于聲明為atomic而且又自己手動(dòng)實(shí)現(xiàn)getter或者setter的屬性，也可以用barrier來(lái)改進(jìn)：

@property (atomic, copy) NSString *someString; 
 
- (NSString *)someString { 
    __block NSString *tempString; 
    dispatch_sync(_syncQueue, ^{ 
        tempString = _someString; 
    }); 
    return tempString; 
} 
 
- (void)setSomeString :(NSString *)someString { 
    dispatch_barrier_async(_syncQueue, ^{ 
        _someString = someString 
        ... 
    } 
} 

在做到atomic的同時(shí)，getter之間還可以并發(fā)執(zhí)行，比直接把setter和getter都放到串行隊(duì)列或者加普通鎖要更高效。

讀者寫者鎖能提升多少效率?

使用讀者寫者鎖一定比所有讀寫都加鎖以及使用串行隊(duì)列要快，但是到底能快多少呢?Dmytro Anokhin在[3]中做了實(shí)驗(yàn)對(duì)比，測(cè)出了分別使用NSLock、GCD barrier和pthread_rwlock時(shí)獲取鎖所需要的平均時(shí)間，實(shí)驗(yàn)樣本數(shù)在100到1000之間，去掉***和***的10%，結(jié)果如下列圖表所示：

3 writers / 10 readers

1 writer / 10 readers

5 writers / 5 readers

10 writers / 1 reader

分析可知：

1. 使用讀者寫者鎖(GCD barrier、pthread_rwlock)，相比單純使用普通鎖(NSLock)，效率有顯著提升;
2. 讀者數(shù)量越多，寫者數(shù)量越少，使用讀者寫者鎖的效率優(yōu)勢(shì)越明顯;
3. 使用GCD barrier和使用pthread_rwlock的效率差異不大。

由于pthread_rwlock不易使用且容易出錯(cuò)，而且GCD barrier和pthread_rwlock對(duì)比性能相當(dāng)，建議使用GCD barrier來(lái)解決iOS開發(fā)中遇到的讀者寫者問(wèn)題。另外，使用GCD還有個(gè)潛在優(yōu)勢(shì)：GCD面向隊(duì)列而非線程，dispatch至某一隊(duì)列的任務(wù)，可能在任一線程上執(zhí)行，這些對(duì)開發(fā)者是透明的，這樣設(shè)計(jì)的好處顯而易見(jiàn)，GCD可以根據(jù)實(shí)際情況從自己管理的線程池中挑選出開銷最小的線程來(lái)執(zhí)行任務(wù)，***程度減小context切換次數(shù)。

何時(shí)使用讀者寫者鎖

需要注意的是，并非所有的多線程讀寫場(chǎng)景都一定是讀者寫者問(wèn)題，使用時(shí)要注意辨別。例如以下YYCache的代碼：

//讀cache 
- (id)objectForKey:(id)key { 
    if (!key) return nil; 
    pthread_mutex_lock(&_lock); 
    _YYLinkedMapNode *node = CFDictionaryGetValue(_lru->_dic, (__bridge const void *)(key)); 
    if (node) { 
        node->_time = CACurrentMediaTime(); 
        [_lru bringNodeToHead:node]; 
    } 
    pthread_mutex_unlock(&_lock); 
    return node ? node->_value : nil; 
} 

//寫cache 
- (void)setObject:(id)object forKey:(id)key withCost:(NSUInteger)cost { 
    if (!key) return; 
    if (!object) { 
        [self removeObjectForKey:key]; 
        return; 
    } 
    pthread_mutex_lock(&_lock); 
    _YYLinkedMapNode *node = CFDictionaryGetValue(_lru->_dic, (__bridge const void *)(key)); 
    NSTimeInterval now = CACurrentMediaTime(); 
    if (node) { 
        _lru->_totalCost -= node->_cost; 
        _lru->_totalCost += cost; 
        node->_cost = cost; 
        node->_time = now; 
        node->_value = object; 
        [_lru bringNodeToHead:node]; 
    } else { 
        node = [_YYLinkedMapNode new]; 
        node->_cost = cost; 
        node->_time = now; 
        node->_key = key; 
        node->_value = object; 
        [_lru insertNodeAtHead:node]; 
    } 
    if (_lru->_totalCost > _costLimit) { 
        dispatch_async(_queue, ^{ 
            [self trimToCost:_costLimit]; 
        }); 
    } 
    if (_lru->_totalCount > _countLimit) { 
        _YYLinkedMapNode *node = [_lru removeTailNode]; 
        if (_lru->_releaseAsynchronously) { 
            dispatch_queue_t queue = _lru->_releaseOnMainThread ? dispatch_get_main_queue() : YYMemoryCacheGetReleaseQueue(); 
            dispatch_async(queue, ^{ 
                [node class]; //hold and release in queue 
            }); 
        } else if (_lru->_releaseOnMainThread && !pthread_main_np()) { 
            dispatch_async(dispatch_get_main_queue(), ^{ 
                [node class]; //hold and release in queue 
            }); 
        } 
    } 
    pthread_mutex_unlock(&_lock); 
} 

這里的cache由于使用了LRU淘汰策略，每次在讀cache的同時(shí)，會(huì)將本次的cache放到數(shù)據(jù)結(jié)構(gòu)的最前面，從而延緩最近使用的cache被淘汰的時(shí)機(jī)，因?yàn)槊看巫x操作的同時(shí)也會(huì)發(fā)生寫操作，所以這里直接使用pthread_mutex互斥鎖，而沒(méi)有使用讀者寫者鎖。

綜上所述，如果你所遇到的多線程讀寫場(chǎng)景符合：

1. 存在單純的讀操作(即讀任務(wù)里沒(méi)有同時(shí)包含寫操作);
2. 讀者數(shù)量較多，而寫者數(shù)量較少。

都應(yīng)該考慮使用讀者寫者鎖來(lái)進(jìn)一步提升并發(fā)率。

注意：

(1)讀者寫者問(wèn)題包含“讀者優(yōu)先”和“寫者優(yōu)先”兩類：前者表示讀線程只要看到有其他讀線程正在訪問(wèn)文件，就可以繼續(xù)作讀訪問(wèn)，寫線程必須等待所有讀線程都不訪問(wèn)時(shí)才能寫文件，即使寫線程可能比一些讀線程更早提出申請(qǐng);而寫者優(yōu)先表示寫線程只要提出申請(qǐng)，再后來(lái)的讀線程就必須等待該寫線程完成。GCD的barrier屬于寫者優(yōu)先的實(shí)現(xiàn)。具體請(qǐng)參考文檔[2]。

(2)串行隊(duì)列上沒(méi)必要使用GCD barrier，應(yīng)該使用dispatch_queue_create建立的并發(fā)隊(duì)列;dispatch_get_global_queue由于是全局共享隊(duì)列，使用barrier達(dá)不到隔離當(dāng)前任務(wù)的效果，會(huì)自動(dòng)降級(jí)為dispatch_sync / dispatch_async。[5]

鎖的粒度(Granularity)

首先看兩段代碼：

代碼段1

@property (atomic, copy) NSString *atomicStr; 
 
//thread A 
atomicSr = @"am on thread A"; 
NSLog(@"%@", atomicStr); 
 
//thread B 
atomicSr = @"am on thread B"; 
NSLog(@"%@", atomicStr); 

代碼段2

- (void)synchronizedAMethod { 
    @synchronized (self) { 
        ... 
    } 
} 
 
- (void)synchronizedBMethod { 
    @synchronized (self) { 
        ... 
    } 
} 
 
- (void)synchronizedCMethod { 
    @synchronized (self) { 
        ... 
    } 
} 

粒度過(guò)小

執(zhí)行代碼段1，在線程A上打印出來(lái)的字符串卻可能是“am on thread B”，原因是雖然atomicStr是原子操作，但是取出atomicStr之后，在執(zhí)行NSLog之前，atomicStr仍然可能會(huì)被線程B修改。因此atomic聲明的屬性，只能保證屬性的get和set是完整的，但是卻不能保證get和set完之后的關(guān)于該屬性的操作是多線程安全的，這就是aomic聲明的屬性不一定能保證多線程安全的原因。

同樣的，不僅僅是atomic聲明的屬性，在開發(fā)中自己加的鎖如果粒度太小，也不能保證線程安全，代碼段1其實(shí)和下面代碼效果一致：

@property (nonatomic, strong) NSLock *lock; 
@property (nonatomic, copy) NSString *atomicStr; 
 
//thread A 
[_lock lock]; 
atomicSr = @"am on thread A"; 
[_lock unlock]; 
NSLog(@"%@", atomicStr); 
 
//thread B 
[_lock lock]; 
atomicSr = @"am on thread B"; 
[_lock unlock]; 
NSLog(@"%@", atomicStr); 

如果想讓程序按照我們的初衷，設(shè)置完atomicStr后打印出來(lái)的就是設(shè)置的值，就需要加大鎖的范圍，將NSLog也包括在臨界區(qū)內(nèi)：

//thread A 
[_lock lock]; 
atomicSr = @"am on thread A"; 
NSLog(@"%@", atomicStr); 
[_lock unlock]; 
 
//thread B 
[_lock lock]; 
atomicSr = @"am on thread B"; 
NSLog(@"%@", atomicStr); 
[_lock unlock]; 

示例代碼很簡(jiǎn)單，很容易看出問(wèn)題所在，但是在實(shí)際開發(fā)中遇到更復(fù)雜些的代碼塊時(shí)，一不小心就可能踏入坑里。因此在設(shè)計(jì)多線程代碼時(shí)，要特別注意代碼之間的邏輯關(guān)系，若后續(xù)代碼依賴于加鎖部分的代碼，那這些后續(xù)代碼也應(yīng)該一并加入鎖中。

粒度過(guò)大

@synchronized關(guān)鍵字會(huì)自動(dòng)根據(jù)傳入對(duì)象創(chuàng)建一個(gè)與之關(guān)聯(lián)的鎖，在代碼塊開始時(shí)自動(dòng)加鎖，并在代碼塊結(jié)束后自動(dòng)解鎖，語(yǔ)法簡(jiǎn)單明了，很方便使用，但是這也導(dǎo)致部分開發(fā)者過(guò)渡依賴于@synchronized關(guān)鍵字，濫用@synchronized(self)。如上述代碼段2中的寫法，在一整個(gè)類文件里，所有加鎖的地方用的都是@synchronized(self)，這就可能會(huì)導(dǎo)致不相關(guān)的線程執(zhí)行時(shí)都要互相等待，原本可以并發(fā)執(zhí)行的任務(wù)不得不串行執(zhí)行。另外使用@synchronized(self)還可能導(dǎo)致死鎖：

//class A 
 
@synchronized (self) { 
    [_sharedLock lock]; 
    NSLog(@"code in class A"); 
    [_sharedLock unlock]; 
} 
 
//class B 
[_sharedLock lock]; 
@synchronized (objectA) { 
    NSLog(@"code in class B"); 
} 
[_sharedLock unlock]; 

原因是因?yàn)閟elf很可能會(huì)被外部對(duì)象訪問(wèn)，被用作key來(lái)生成一鎖，類似上述代碼中的@synchronized (objectA)。兩個(gè)公共鎖交替使用的場(chǎng)景就容易出現(xiàn)死鎖。所以正確的做法是傳入一個(gè)類內(nèi)部維護(hù)的NSObject對(duì)象，而且這個(gè)對(duì)象是對(duì)外不可見(jiàn)的[2]。

因此，不相關(guān)的多線程代碼，要設(shè)置不同的鎖，一個(gè)鎖只管一個(gè)臨界區(qū)。除此之外，還有種常見(jiàn)的錯(cuò)誤做法會(huì)導(dǎo)致并發(fā)效率下降：

//thread A 
[_lock lock]; 
atomicSr = @"am on thread A"; 
NSLog(@"%@", atomicStr); 
//do some other tasks which are none of business with atomicStr; 
for (int i = 0; i < 100000; i ++) { 
    sleep(5); 
} 
[_lock unlock]; 
 
//thread B 
[_lock lock]; 
atomicSr = @"am on thread B"; 
NSLog(@"%@", atomicStr); 
//do some other tasks which are none of business with atomicStr; 
for (int i = 0; i < 100000; i ++) { 
    sleep(5); 
} 
[_lock unlock]; 

即在臨界區(qū)內(nèi)包含了與當(dāng)前加鎖對(duì)象無(wú)關(guān)的任務(wù)，實(shí)際應(yīng)用中，需要我們尤其注意臨界區(qū)內(nèi)的每一個(gè)函數(shù)，因?yàn)槠鋬?nèi)部實(shí)現(xiàn)可能調(diào)用了耗時(shí)且無(wú)關(guān)的任務(wù)。

遞歸鎖(Recursive lock)

相比較上述提到的@synchronized(self)，下面這種情形引起的死鎖更加常見(jiàn)：

@property (nonatomic,strong) NSLock *lock; 
 
_lock = [[NSLock alloc] init]; 
 
- (void)synchronizedAMethod { 
    [_lock lock]; 
    //do some tasks 
    [self synchronizedBMethod]; 
    [_lock unlock]; 
} 
 
- (void)synchronizedBMethod { 
    [_lock lock]; 
    //do some tasks 
    [_lock unlock]; 
} 

A方法已獲取鎖后，再調(diào)用B方法，就會(huì)觸發(fā)死鎖，B方法在等待A方法執(zhí)行完成釋放鎖后才能繼續(xù)執(zhí)行，而A方法執(zhí)行完成的前提是執(zhí)行完B方法。實(shí)際開發(fā)中，可能發(fā)生死鎖的情形往往隱蔽在方法的層層調(diào)用中。因此建議在不能確定是否會(huì)產(chǎn)生死鎖時(shí)，***使用遞歸鎖。更保守一點(diǎn)的做法是不論何時(shí)都使用遞歸鎖，因?yàn)楹茈y保證以后的代碼會(huì)不會(huì)在同一線程上多次加鎖。

遞歸鎖允許同一個(gè)線程在未釋放其擁有的鎖時(shí)反復(fù)對(duì)該鎖進(jìn)行加鎖操作，內(nèi)部通過(guò)一個(gè)計(jì)數(shù)器來(lái)實(shí)現(xiàn)。除了NSRecursiveLock，也可以使用性能更佳的pthread_mutex_lock，初始化時(shí)參數(shù)設(shè)置為PTHREAD_MUTEX_RECURSIVE即可：

pthread_mutexattr_t attr; 
pthread_mutexattr_init (&attr); 
pthread_mutexattr_settype (&attr, PTHREAD_MUTEX_RECURSIVE); 
pthread_mutex_init (&_lock, &attr); 
pthread_mutexattr_destroy (&attr); 

值得注意的是，@synchronized內(nèi)部使用的也是遞歸鎖：

// Begin synchronizing on 'obj'.  
// Allocates recursive mutex associated with 'obj' if needed. 
// Returns OBJC_SYNC_SUCCESS once lock is acquired.   
int objc_sync_enter(id obj) 
{ 
    int result = OBJC_SYNC_SUCCESS; 
 
    if (obj) { 
        SyncData* data = id2data(obj, ACQUIRE); 
        assert(data); 
        data->mutex.lock(); 
    } else { 
        // @synchronized(nil) does nothing 
        if (DebugNilSync) { 
            _objc_inform("NIL SYNC DEBUG: @synchronized(nil); set a breakpoint on objc_sync_nil to debug"); 
        } 
        objc_sync_nil(); 
    } 
 
    return result; 
} 

總結(jié)

想寫出高效、安全的多線程代碼，只是熟悉GCD、@synchronized、NSLock這幾個(gè)API是不夠的，還需要了解更多API背后的知識(shí)，深刻理解臨界區(qū)的概念、理清各個(gè)任務(wù)之間的時(shí)序關(guān)系是必要條件。