synchronized分析

我們先來(lái)看個(gè)題目：

- (void)lg_testSaleTicket{
    dispatch_async(dispatch_get_global_queue(0, 0), ^{
        for (int i = 0; i < 5; i++) {
            [self saleTicket];
        }
    });
    
    dispatch_async(dispatch_get_global_queue(0, 0), ^{
        for (int i = 0; i < 5; i++) {
            [self saleTicket];
        }
    });
    
    dispatch_async(dispatch_get_global_queue(0, 0), ^{
        for (int i = 0; i < 10; i++) {
            [self saleTicket];
        }
    });
}

- (void)saleTicket{
    if (self.ticketCount > 0) {
        self.ticketCount--;
        sleep(0.1);
        NSLog(@"當(dāng)前余票還剩：%ld張",self.ticketCount);
    }else{
        NSLog(@"當(dāng)前車票已售罄");
    }
}

然后我們調(diào)用上面的方法

self.ticketCount = 20;
[self lg_testSaleTicket];

請(qǐng)問(wèn)上面的代碼設(shè)計(jì)是否有問(wèn)題呢？
當(dāng)然有問(wèn)題，會(huì)存在多個(gè)線程操作一個(gè)數(shù)據(jù)ticketCount，導(dǎo)致數(shù)據(jù)不安全的問(wèn)題。執(zhí)行完成后剩余的票數(shù)可能不會(huì)為0。
既然是多線程導(dǎo)致的數(shù)據(jù)不安全問(wèn)題，我們就可以加鎖進(jìn)行解決。

- (void)saleTicket{
    // 枷鎖 - 線程安全
    @synchronized (self) {
        if (self.ticketCount > 0) {
            self.ticketCount--;
            sleep(0.1);
            NSLog(@"當(dāng)前余票還剩：%ld張",self.ticketCount);
        }else{
            NSLog(@"當(dāng)前車票已售罄");
        }
    }
}

我們對(duì)賣票的操作部分加上了@synchronized，這樣同時(shí)只能有一個(gè)線程操作ticketCount，從而保證了數(shù)據(jù)的安全。
下面我們來(lái)探究下@synchronized。

appDelegateClassName = NSStringFromClass([AppDelegate class]);
@synchronized (appDelegateClassName) {
}

在synchronized的地方打上斷點(diǎn)，然后匯編調(diào)試。

image.png

在synchronized的匯編調(diào)試代碼中，我們看有objc_sync_enter和objc_sync_exit成對(duì)的出現(xiàn)。所以這一對(duì)函數(shù)應(yīng)該是和synchronized的底層實(shí)現(xiàn)相關(guān)的。
然后我們就可以通過(guò)符號(hào)斷點(diǎn)，針對(duì)objc_sync_enter打個(gè)符號(hào)斷點(diǎn)

objc_sync_enter符號(hào)斷點(diǎn)

下載objc源碼，然后搜索objc_sync_enter

// 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;
}

synchronizing是種互斥鎖。首先判斷objc，如果不存在的話走objc_sync_nil()，也就是什么都不做。所以在使用@synchronized(obj)進(jìn)行加鎖的時(shí)候，如果obj為nil，就是無(wú)效的，不會(huì)進(jìn)行加鎖。

下面我們看下objc不為空的情況：
構(gòu)建了SyncData，看下SyncData的結(jié)構(gòu)

typedef struct alignas(CacheLineSize) SyncData {
    struct SyncData* nextData;
    DisguisedPtr<objc_object> object;
    int32_t threadCount;  // number of THREADS using this block
    recursive_mutex_t mutex;
} SyncData;

里面有個(gè)nextData，應(yīng)該是指向了下一個(gè)節(jié)點(diǎn)。所以好多這樣的節(jié)點(diǎn)組成了一個(gè)鏈表似的結(jié)構(gòu)；里面還有個(gè)遞歸鎖mutex(遞歸鎖屬于互斥鎖的一種)。

static SyncData* id2data(id object, enum usage why)
{
    spinlock_t *lockp = &LOCK_FOR_OBJ(object);
    // ....... 
}

在id2data函數(shù)中通過(guò)LOCK_FOR_OBJ函數(shù)獲取到lockp，LOCK_FOR_OBJ函數(shù)的定義如下

#define LOCK_FOR_OBJ(obj) sDataLists[obj].lock
static StripedMap<SyncList> sDataLists;

可以看到sDataLists實(shí)際上是一個(gè)哈希表，表中存在一個(gè)個(gè)的SyncList對(duì)象，SyncList對(duì)象的結(jié)構(gòu)中有data和lock。

struct SyncList {
    SyncData *data;
    spinlock_t lock;

    constexpr SyncList() : data(nil), lock(fork_unsafe_lock) { }
};

@synchronized底層是封裝的互斥鎖pThread。

synchronized使用注意點(diǎn)

下面代碼可以正常運(yùn)行嗎？

- (void)lg_crash{
    for (int i = 0; i < 200000; i++) {
        dispatch_async(dispatch_get_global_queue(0, 0), ^{
            _testArray = [NSMutableArray array];
        });
    }
}

執(zhí)行這段代碼會(huì)導(dǎo)致野指針crash。
GCD里面的_testArray = [NSMutableArray array];這句代碼是在創(chuàng)建新的Array賦值給_testArray，然后釋放了舊值。如果此時(shí)多個(gè)線程同時(shí)暫存了舊值，然后就會(huì)導(dǎo)致多次釋放同一個(gè)舊值，從而產(chǎn)生野指針崩潰。
我們可以進(jìn)行加鎖處理。像下面的這樣加鎖處理可以嗎？

- (void)lg_crash{
    for (int i = 0; i < 200000; i++) {
        dispatch_async(dispatch_get_global_queue(0, 0), ^{
            @synchronized (_testArray) {
                _testArray = [NSMutableArray array];
            }
        });
    }
}

答案是會(huì)產(chǎn)生同樣的野指針crash。因?yàn)樵谶^(guò)程中_testArray可能為空，使用@synchronized鎖的對(duì)象如果為空的話，相當(dāng)于不鎖。所以會(huì)得到同樣的crash。此時(shí)我們可以將鎖的對(duì)象_testArray換成self，這樣就可以解決問(wèn)題。但是@synchronized底層需要對(duì)哈希表進(jìn)行處理，過(guò)程比較復(fù)雜，所以效率低。這里我們可以使用NSLock來(lái)進(jìn)行加鎖處理。

- (void)lg_crash{
    NSLock *lock = [[NSLock alloc] init];
    for (int i = 0; i < 200000; i++) {
        dispatch_async(dispatch_get_global_queue(0, 0), ^{
            [lock lock];
            _testArray = [NSMutableArray array];
            [lock unlock];
        });
    }
}

NSLock分析

下面的代碼可以正常執(zhí)行嗎？

NSLock *lock = [[NSLock alloc] init];
dispatch_async(dispatch_get_global_queue(0, 0), ^{
    static void (^testMethod)(int);
    testMethod = ^(int value){
        [lock lock];
        if (value > 0) {
            NSLog(@"current value = %d",value);
            testMethod(value - 1);
        }
        [lock unlock];
    };
    testMethod(10);
})

答案是只打印出一個(gè)10，就會(huì)卡死。
因?yàn)檫f歸調(diào)用了testMethod，就會(huì)多次進(jìn)行l(wèi)ock加鎖，在一個(gè)lock鎖定的區(qū)域內(nèi)遞歸調(diào)用再次進(jìn)行加鎖，就會(huì)導(dǎo)致堵塞。
因?yàn)槭沁f歸調(diào)用，此時(shí)我們應(yīng)該講NSLock換成遞歸鎖NSRecursiveLock，就能正常的打印出10 9 8 7 6 5 4 3 2 1 了。

我們?cè)谏厦娲a的最外層再加一個(gè)for循環(huán)，還可以正常執(zhí)行嗎？

NSRecursiveLock *lock = [[NSRecursiveLock alloc] init];
for (int i= 0; i<100; i++) {
    dispatch_async(dispatch_get_global_queue(0, 0), ^{
        static void (^testMethod)(int);
        testMethod = ^(int value){
            [lock lock];
            if (value > 0) {
              NSLog(@"current value = %d",value);
              testMethod(value - 1);
            }
            [lock unlock];
        };
        testMethod(10);
    });
}

這樣就會(huì)導(dǎo)致死鎖的問(wèn)題。多個(gè)線程進(jìn)行加鎖，互相等待，導(dǎo)致死鎖。此時(shí)我們只需要將遞歸鎖換成@synchronized就可以解決死鎖問(wèn)題了。因?yàn)锧synchronized的底層的實(shí)現(xiàn)，如果已經(jīng)鎖過(guò)一次了就會(huì)從緩存中取，而不會(huì)再次加鎖了。
總結(jié)：普通的線程安全可以使用NSLock；如果存在遞歸調(diào)用，使用NSRecursiveLock；如果內(nèi)部存在遞歸，外部存在循環(huán)或者有其他線程影響，使用@synchronized。

條件鎖：NSCondition

調(diào)用下面的lg_testConditon方法，會(huì)有問(wèn)題嗎？

- (void)lg_testConditon{
    //創(chuàng)建生產(chǎn)-消費(fèi)者
    for (int i = 0; i < 50; i++) {
        dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_HIGH, 0), ^{
            [self lg_producer];
        });
        dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_HIGH, 0), ^{
            [self lg_consumer];
        });
    }
}

- (void)lg_producer{
    self.ticketCount = self.ticketCount + 1;
    NSLog(@"生產(chǎn)一個(gè) 現(xiàn)有 count %zd",self.ticketCount);
}

- (void)lg_consumer{
    while (self.ticketCount == 0) {
        NSLog(@"等待 count %zd",self.ticketCount);
    }
    
    //注意消費(fèi)行為，要在等待條件判斷之后
    self.ticketCount -= 1;
    NSLog(@"消費(fèi)一個(gè) 還剩 count %zd ",self.ticketCount);
}

上面的代碼因?yàn)樵诙嗑€程中，不能保證數(shù)據(jù)安全。我們需要加鎖處理？這里NSCondition就最合適了。使用NSCondition當(dāng)消費(fèi)到ticketCount為0的時(shí)候，調(diào)用wait等待。當(dāng)生產(chǎn)一個(gè)ticket后，調(diào)用signal發(fā)送信號(hào)，讓等待的可以繼續(xù)執(zhí)行。代碼實(shí)現(xiàn)如下：

- (void)lg_producer{
    [_testCondition lock];
    self.ticketCount = self.ticketCount + 1;
    NSLog(@"生產(chǎn)一個(gè) 現(xiàn)有 count %zd",self.ticketCount);
    [_testCondition signal];
    [_testCondition unlock];
}

- (void)lg_consumer{
    // 線程安全
    [_testCondition lock];
    while (self.ticketCount == 0) {
        NSLog(@"等待 count %zd",self.ticketCount);
        // 保證正常流程
        [_testCondition wait];
    }
    //注意消費(fèi)行為，要在等待條件判斷之后
    self.ticketCount -= 1;
    NSLog(@"消費(fèi)一個(gè) 還剩 count %zd ",self.ticketCount);
    [_testCondition unlock];
}

首先鎖住生產(chǎn)和消費(fèi)的代碼，然后在消費(fèi)的時(shí)候如果發(fā)現(xiàn)ticketCount為0，就wait等待。生產(chǎn)后發(fā)送signal，讓等待的繼續(xù)執(zhí)行消費(fèi)。

條件鎖：NSConditionLock

// 信號(hào)量
NSConditionLock *conditionLock = [[NSConditionLock alloc] initWithCondition:2];
dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_HIGH, 0), ^{
   [conditionLock lockWhenCondition:1];
   NSLog(@"線程 1");
   [conditionLock unlockWithCondition:0];
});

dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_LOW, 0), ^{
   [conditionLock lockWhenCondition:2];
   NSLog(@"線程 2");  
   [conditionLock unlockWithCondition:1];
});
dispatch_async(dispatch_get_global_queue(0, 0), ^{
   [conditionLock lock];
   NSLog(@"線程 3");
   [conditionLock unlock];
});

首先創(chuàng)建一個(gè)NSConditionLock條件鎖，并且設(shè)置condition為2。
[conditionLock lockWhenCondition:1];的意思是如果此時(shí)的condition為1，并且沒(méi)有其他線程獲取鎖，那么就可以獲取鎖執(zhí)行下面的代碼。[conditionLock unlockWithCondition:0];的意思是釋放鎖，并且將條件置為0。[conditionLock lock];的意思是不受condition條件的影響。
了解了NSConditionLock后，我們可以知道，線程2肯定是在線程1之前執(zhí)行。

下面我們來(lái)使用匯編來(lái)探索一下NSCondition的實(shí)現(xiàn)，首先在[conditionLock lockWhenCondition:1];的地方打上斷點(diǎn)，然后開(kāi)啟匯編調(diào)試

image.png

image.png

image.png

方法的執(zhí)行者是NSConditionLock，方法名稱為lockWhenCondition:beforeDate:。我們?cè)谔O果的官方文檔也找到了這個(gè)方法。我們繼續(xù)打符號(hào)斷點(diǎn)追蹤這個(gè)方法的實(shí)現(xiàn)。

image.png

image.png

image.png