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

讀者寫者問題(Readers-writers problem)

先看下讀者寫者問題的描述：

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

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

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

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

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

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

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

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

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

pthread_rwlock_t

這是一個古老的C語言層面的函數(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) 

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

GCD barrier

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

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

//實(shí)現(xià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)一個簡單的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í)行讀操作，同時又有效地隔離了寫操作，兼顧了安全和效率。

對于聲明為atomic而且又自己手動實(shí)現(xiàn)getter或者setter的屬性，也可以用barrier來改進(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的同時，getter之間還可以并發(fā)執(zhí)行，比直接把setter和getter都放到串行隊列或者加普通鎖要更高效。

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

使用讀者寫者鎖一定比所有讀寫都加鎖以及使用串行隊列要快，但是到底能快多少呢?Dmytro Anokhin在[3]中做了實(shí)驗(yàn)對比，測出了分別使用NSLock、GCD barrier和pthread_rwlock時獲取鎖所需要的平均時間，實(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)勢越明顯;
3. 使用GCD barrier和使用pthread_rwlock的效率差異不大。

由于pthread_rwlock不易使用且容易出錯，而且GCD barrier和pthread_rwlock對比性能相當(dāng)，建議使用GCD barrier來解決iOS開發(fā)中遇到的讀者寫者問題。另外，使用GCD還有個潛在優(yōu)勢：GCD面向隊列而非線程，dispatch至某一隊列的任務(wù)，可能在任一線程上執(zhí)行，這些對開發(fā)者是透明的，這樣設(shè)計的好處顯而易見，GCD可以根據(jù)實(shí)際情況從自己管理的線程池中挑選出開銷最小的線程來執(zhí)行任務(wù)，***程度減小context切換次數(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的同時，會將本次的cache放到數(shù)據(jù)結(jié)構(gòu)的最前面，從而延緩最近使用的cache被淘汰的時機(jī)，因?yàn)槊看巫x操作的同時也會發(fā)生寫操作，所以這里直接使用pthread_mutex互斥鎖，而沒有使用讀者寫者鎖。

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

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

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

注意：

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

(2)串行隊列上沒必要使用GCD barrier，應(yīng)該使用dispatch_queue_create建立的并發(fā)隊列;dispatch_get_global_queue由于是全局共享隊列，使用barrier達(dá)不到隔離當(dāng)前任務(wù)的效果，會自動降級為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) { 
        ... 
    } 
} 

粒度過小

執(zhí)行代碼段1，在線程A上打印出來的字符串卻可能是“am on thread B”，原因是雖然atomicStr是原子操作，但是取出atomicStr之后，在執(zhí)行NSLog之前，atomicStr仍然可能會被線程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后打印出來的就是設(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]; 

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

粒度過大

@synchronized關(guān)鍵字會自動根據(jù)傳入對象創(chuàng)建一個與之關(guān)聯(lián)的鎖，在代碼塊開始時自動加鎖，并在代碼塊結(jié)束后自動解鎖，語法簡單明了，很方便使用，但是這也導(dǎo)致部分開發(fā)者過渡依賴于@synchronized關(guān)鍵字，濫用@synchronized(self)。如上述代碼段2中的寫法，在一整個類文件里，所有加鎖的地方用的都是@synchronized(self)，這就可能會導(dǎo)致不相關(guān)的線程執(zhí)行時都要互相等待，原本可以并發(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很可能會被外部對象訪問，被用作key來生成一鎖，類似上述代碼中的@synchronized (objectA)。兩個公共鎖交替使用的場景就容易出現(xiàn)死鎖。所以正確的做法是傳入一個類內(nèi)部維護(hù)的NSObject對象，而且這個對象是對外不可見的[2]。

因此，不相關(guān)的多線程代碼，要設(shè)置不同的鎖，一個鎖只管一個臨界區(qū)。除此之外，還有種常見的錯誤做法會導(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)前加鎖對象無關(guān)的任務(wù)，實(shí)際應(yīng)用中，需要我們尤其注意臨界區(qū)內(nèi)的每一個函數(shù)，因?yàn)槠鋬?nèi)部實(shí)現(xiàn)可能調(diào)用了耗時且無關(guān)的任務(wù)。

遞歸鎖(Recursive lock)

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

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

遞歸鎖允許同一個線程在未釋放其擁有的鎖時反復(fù)對該鎖進(jìn)行加鎖操作，內(nèi)部通過一個計數(shù)器來實(shí)現(xiàn)。除了NSRecursiveLock，也可以使用性能更佳的pthread_mutex_lock，初始化時參數(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這幾個API是不夠的，還需要了解更多API背后的知識，深刻理解臨界區(qū)的概念、理清各個任務(wù)之間的時序關(guān)系是必要條件。