深入理解 iOS 开发中的锁

澳门新葡亰平台官网 1

摘要

本文的目的不是介绍 iOS
中各种锁如何使用,一方面笔者没有大量的实战经验,另一方面这样的文章相当多,比如 iOS中保证线程安全的几种方式与性能对比、iOS
常见知识点(三):Lock。本文也不会详细介绍锁的具体实现原理,这会涉及到太多相关知识,笔者不敢误人子弟。

本文要做的就是简单的分析 iOS
开发中常见的几种锁如何实现,以及优缺点是什么,为什么会有性能上的差距,最终会简单的介绍锁的底层实现原理。水平有限,如果不慎有误,欢迎交流指正。同时建议读者在阅读本文以前,对
OC 中各种锁的使用方法先有大概的认识。

在 ibireme 的 不再安全的
OSSpinLock 一文中,有一张图片简单的比较了各种锁的加解锁性能:

澳门新葡亰平台官网 1

本文会按照从上至下(速度由快至慢)的顺序分析每个锁的实现原理。需要说明的是,加解锁速度不表示锁的效率,只表示加解锁操作在执行时的复杂程度,下文会通过具体的例子来解释。

iOS线程安全的锁与性能对比

OSSpinLock

上述文章中已经介绍了 OSSpinLock
不再安全,主要原因发生在低优先级线程拿到锁时,高优先级线程进入忙等(busy-wait)状态,消耗大量
CPU 时间,从而导致低优先级线程拿不到 CPU
时间,也就无法完成任务并释放锁。这种问题被称为优先级反转。

为什么忙等会导致低优先级线程拿不到时间片?这还得从操作系统的线程调度说起。

现代操作系统在管理普通线程时,通常采用时间片轮转算法(Round Robin,简称
RR)。每个线程会被分配一段时间片(quantum),通常在 10-100
毫秒左右。当线程用完属于自己的时间片以后,就会被操作系统挂起,放入等待队列中,直到下一次被分配时间片。

一、锁的基本使用方法

自旋锁的实现原理

自旋锁的目的是为了确保临界区只有一个线程可以访问,它的使用可以用下面这段伪代码来描述:

do {  
    Acquire Lock
        Critical section  // 临界区
    Release Lock
        Reminder section // 不需要锁保护的代码
}

在 Acquire Lock 这一步,我们申请加锁,目的是为了保护临界区(Critical
Section) 中的代码不会被多个线程执行。

自旋锁的实现思路很简单,理论上来说只要定义一个全局变量,用来表示锁的可用情况即可,伪代码如下:

bool lock = false; // 一开始没有锁上,任何线程都可以申请锁  
do {  
    while(lock); // 如果 lock 为 true 就一直死循环,相当于申请锁
    lock = true; // 挂上锁,这样别的线程就无法获得锁
        Critical section  // 临界区
    lock = false; // 相当于释放锁,这样别的线程可以进入临界区
        Reminder section // 不需要锁保护的代码        
}

注释写得很清楚,就不再逐行分析了。可惜这段代码存在一个问题:
如果一开始有多个线程同时执行 while
循环,他们都不会在这里卡住,而是继续执行,这样就无法保证锁的可靠性了。解决思路也很简单,只要确保申请锁的过程是原子操作即可。

1.1、@synchronized

这是我们最熟悉的枷锁方式,用起来也比较简单。使用时@synchronized后面需要紧跟一个OC对象,它实际上是把这个对象当做锁来使用,作为该锁的唯一标识,只有当标识相同时,才会满足互斥。底层原理是当你调用sychronized的每个对象,runtime都会为其分配一个递归锁并储存在哈希表中。@synchronized原理

 NSObject *obj = [[NSObject alloc] init];
    dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
        @synchronized (obj) {
            NSLog(@"需要线程同步的操作1  开始");
            sleep(3);
            NSLog(@"需要线程同步的操作1  结束");
        }
    });

    dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
        sleep(1);
        @synchronized (obj) {
            NSLog(@"需要线程同步的操作 2 ");
        }
    });

执行结果很显然,等第一个线程结束后,才执行到第二个线程中

原子操作

狭义上的原子操作表示一条不可打断的操作,也就是说线程在执行操作过程中,不会被操作系统挂起,而是一定会执行完。在单处理器环境下,一条汇编指令显然是原子操作,因为中断也要通过指令来实现。

然而在多处理器的情况下,能够被多个处理器同时执行的操作任然算不上原子操作。因此,真正的原子操作必须由硬件提供支持,比如
x86 平台上如果在指令前面加上 “LOCK”
前缀,对应的机器码在执行时会把总线锁住,使得其他
CPU不能再执行相同操作,从而从硬件层面确保了操作的原子性。

这些非常底层的概念无需完全掌握,我们只要知道上述申请锁的过程,可以用一个原子性操作test_and_set 来完成,它用伪代码可以这样表示:

bool test_and_set (bool *target) {  
    bool rv = *target; 
    *target = TRUE; 
    return rv;
}

这段代码的作用是把 target 的值设置为
1,并返回原来的值。当然,在具体实现时,它通过一个原子性的指令来完成。

1.2、dispatch_semaphore

dispatch_semaphore是GCD用来同步的一种方式,与他相关的共有三个函数,分别是dispatch_semaphore_create,dispatch_semaphore_signal,dispatch_semaphore_wait。

这个函数会使传入的信号量dsema的值减1;如果dsema信号量的值大于0,该函数所处线程就继续执行下面的语句,并且将信号量的值减1;如果desema的值为0,那么这个函数就阻塞当前线程等待timeout(注意timeout的类型为dispatch_time_t,不能直接传入整形或float型数),如果等待的期间desema的值被dispatch_semaphore_signal函数加1了,且该函数(即dispatch_semaphore_wait)所处线程获得了信号量,那么就继续向下执行并将信号量减1。如果等待期间没有获取到信号量或者信号量的值一直为0,那么等到timeout时,其所处线程自动执行其后语句。

dispatch_semaphore 是信号量,但当信号总量设为 1
时也可以当作锁来。在没有等待情况出现时,它的性能比 pthread_mutex
还要高,但一旦有等待情况出现时,性能就会下降许多。相对于 OSSpinLock
来说,它的优势在于等待时不会消耗 CPU 资源。

如上的代码,如果超时时间overTime设置成>2,可完成同步操作。如果overTime<2的话,在线程1还没有执行完成的情况下,此时超时了,将自动执行下面的代码。

 dispatch_semaphore_t signal = dispatch_semaphore_create(1);  // 信号量为1
    dispatch_time_t overTime = dispatch_time(DISPATCH_TIME_NOW, 3 * NSEC_PER_SEC);

    dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{

        dispatch_semaphore_wait(signal, overTime);  // 信号量减1 ; overtime 超时时间 过了超时时间不会等待
        NSLog(@"需要线程同步的操作1 开始");
        sleep(2);
        NSLog(@"需要线程同步的操作1 结束");
        dispatch_semaphore_signal(signal);
    });

    dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
        sleep(1);
        dispatch_semaphore_wait(signal, overTime);
        NSLog(@"需要线程同步的操作2  ");
        dispatch_semaphore_signal(signal);
    });

上面代码的执行结果:

2017-09-05 14:16:06.691 demo_lock[16669:271470] 需要线程同步的操作1
开始
2017-09-05 14:16:08.693 demo_lock[16669:271470] 需要线程同步的操作1
结束
2017-09-05 14:16:09.696 demo_lock[16669:271453] 需要线程同步的操作2

把overTime设置成<=2s 执行结果为 :

2017-09-05 14:21:18.422 demo_lock[16746:280666] 需要线程同步的操作1
开始
2017-09-05 14:21:20.428 demo_lock[16746:280667] 需要线程同步的操作2
2017-09-05 14:21:20.428 demo_lock[16746:280666] 需要线程同步的操作1
结束

自旋锁的总结

至此,自旋锁的实现原理就很清楚了:

bool lock = false; // 一开始没有锁上,任何线程都可以申请锁  
do {  
    while(test_and_set(&lock); // test_and_set 是一个原子操作
        Critical section  // 临界区
    lock = false; // 相当于释放锁,这样别的线程可以进入临界区
        Reminder section // 不需要锁保护的代码        
}

如果临界区的执行时间过长,使用自旋锁不是个好主意。之前我们介绍过时间片轮转算法,线程在多种情况下会退出自己的时间片。其中一种是用完了时间片的时间,被操作系统强制抢占。除此以外,当线程进行
I/O 操作,或进入睡眠状态时,都会主动让出时间片。显然在 while
循环中,线程处于忙等状态,白白浪费 CPU
时间,最终因为超时被操作系统抢占时间片。如果临界区执行时间较长,比如是文件读写,这种忙等是毫无必要的。

1.3、NSLock

NSLock 遵循 NSLocking 协议,lock 方法是加锁,unlock 是解锁,tryLock
是尝试加锁,如果失败的话返回 NO,lockBeforeDate:
是在指定Date之前尝试加锁,如果在指定时间之前都不能加锁,则返回NO。

  NSLock *lock = [[NSLock alloc] init];
    dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
        [lock lock];
        NSLog(@"线程1  开始");
        sleep(2);
        NSLog(@"线程1  结束");
        [lock unlock];
    });

    dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
        sleep(1); // 保证线程二的代码后执行
        [lock lock]; // lock的第一秒 一直轮询请求枷锁, 后面阶段会进入waiting状态
        NSLog(@"线程2 开始");
        [lock unlock];
    });

控制台输出结果:
2017-09-05 15:08:33.782 demo_lock[17411:327912] 线程1 开始
2017-09-05 15:08:35.786 demo_lock[17411:327912] 线程1 结束
2017-09-05 15:08:35.786 demo_lock[17411:327911] 线程2 开始

线程 1 中的 lock 锁上了,所以线程 2 中的 lock 加锁失败,阻塞线程 2,但 2
s 后线程 1 中的 lock 解锁,线程 2 就立即加锁成功,执行线程 2
中的后续代码。

查到的资料显示互斥锁会使得线程阻塞,阻塞的过程又分两个阶段,第一阶段是会先空转,可以理解成跑一个
while 循环,不断地去申请加锁,在空转一定时间之后,线程会进入 waiting
状态,此时线程就不占用CPU资源了,等锁可用的时候,这个线程会立即被唤醒。

所以如果将上面线程 1 中的 sleep(2); 改成 sleep(10); 输出的结果会变成:

2017-09-05 15:09:30.271 demo_lock[17431:329738] 线程1 开始
2017-09-05 15:09:40.277 demo_lock[17431:329738] 线程1 结束
2017-09-05 15:09:40.277 demo_lock[17431:329740] 线程2 开始

NSLock只是在内部封装了一个pthread_mutex,
属性为PTHREAD_MUTEX_ERRORCHECK

信号量

之前我在 介绍 GCD
底层实现的文章 中简单描述了信号量 dispatch_semaphore_t 的实现原理,它最终会调用到 sem_wait 方法,这个方法在
glibc 中被实现如下:

int sem_wait (sem_t *sem) {  
  int *futex = (int *) sem;
  if (atomic_decrement_if_positive (futex) > 0)
    return 0;
  int err = lll_futex_wait (futex, 0);
    return -1;
)

首先会把信号量的值减一,并判断是否大于零。如果大于零,说明不用等待,所以立刻返回。具体的等待操作在 lll_futex_wait 函数中实现,lll 是
low level lock
的简称。这个函数通过汇编代码实现,调用到 SYS_futex 这个系统调用,使线程进入睡眠状态,主动让出时间片,这个函数在互斥锁的实现中,也有可能被用到。

主动让出时间片并不总是代表效率高。让出时间片会导致操作系统切换到另一个线程,这种上下文切换通常需要
10
微秒左右,而且至少需要两次切换。如果等待时间很短,比如只有几个微秒,忙等就比线程睡眠更高效。

可以看到,自旋锁和信号量的实现都非常简单,这也是两者的加解锁耗时分别排在第一和第二的原因。再次强调,加解锁耗时不能准确反应出锁的效率(比如时间片切换就无法发生),它只能从一定程度上衡量锁的实现复杂程度。

1.4、NSRecursiveLock递归锁

递归锁也是通过 pthread_mutex_lock
函数来实现,在函数内部会判断锁的类型。NSRecursiveLock 与 NSLock
的区别在于内部封装的 pthread_mutex_t 对象的类型不同,前者的类型为
PTHREAD_MUTEX_RECURSIVE。

//    NSLock *lock = [[NSLock alloc] init];
    NSRecursiveLock *lock = [[NSRecursiveLock alloc] init]; 
    dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
        static void  (^RecursiveMethod)(int);
        RecursiveMethod = ^(int value) {
            [lock lock];
            if (value > 0) {
                NSLog(@"value = %d", value);
                sleep(1);
                RecursiveMethod(value - 1);
            }
            [lock unlock];
        };
        RecursiveMethod(5);
    });

如果是NSLock这段代码是一个典型的死锁情况。在我们的线程中,RecursiveMethod是递归调用的。所以每次进入这个block时,都会去加一次锁,而从第二次开始,由于锁已经被使用了且没有解锁,所以它需要等待锁被解除,这样就导致了死锁,线程被阻塞住了。调试器中会输出如下信息:

2017-09-05 16:01:29.158 demo_lock[18197:370719] value = 5
2017-09-05 16:01:30.161 demo_lock[18197:370719] *** -[NSLock
lock]: deadlock (<NSLock: 0x6000000cd120> ‘(null)’)
2017-09-05 16:01:30.161 demo_lock[18197:370719] *** Break on
_NSLockError() to debug.

在这种情况下,我们就可以使用NSRecursiveLock。它可以允许同一线程多次加锁,而不会造成死锁。递归锁会跟踪它被lock的次数。每次成功的lock都必须平衡调用unlock操作。只有所有达到这种平衡,锁最后才能被释放,以供其它线程使用。

我们将NSLock替换为NSRecursiveLock,上面代码则会正确执行

2017-09-05 16:02:56.521 demo_lock[18223:372981] value = 5
2017-09-05 16:02:57.526 demo_lock[18223:372981] value = 4
2017-09-05 16:02:58.531 demo_澳门新葡亰平台官网,lock[18223:372981] value = 3
2017-09-05 16:02:59.531 demo_lock[18223:372981] value = 2
2017-09-05 16:03:00.537 demo_lock[18223:372981] value = 1

pthread_mutex

pthread 表示 POSIX thread,定义了一组跨平台的线程相关的
API,pthread_mutex
表示互斥锁。互斥锁的实现原理与信号量非常相似,不是使用忙等,而是阻塞线程并睡眠,需要进行上下文切换。

互斥锁的常见用法如下:

pthread_mutexattr_t attr;  
pthread_mutexattr_init(&attr);  
pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_NORMAL);  // 定义锁的属性

pthread_mutex_t mutex;  
pthread_mutex_init(&mutex, &attr) // 创建锁

pthread_mutex_lock(&mutex); // 申请锁  
    // 临界区
pthread_mutex_unlock(&mutex); // 释放锁

对于 pthread_mutex
来说,它的用法和之前没有太大的改变,比较重要的是锁的类型,可以有PTHREAD_MUTEX_NORMALPTHREAD_MUTEX_ERRORCHECKPTHREAD_MUTEX_RECURSIVE 等等,具体的特性就不做解释了,网上有很多相关资料。

一般情况下,一个线程只能申请一次锁,也只能在获得锁的情况下才能释放锁,多次申请锁或释放未获得的锁都会导致崩溃。假设在已经获得锁的情况下再次申请锁,线程会因为等待锁的释放而进入睡眠状态,因此就不可能再释放锁,从而导致死锁。

然而这种情况经常会发生,比如某个函数申请了锁,在临界区内又递归调用了自己。辛运的是pthread_mutex 支持递归锁,也就是允许一个线程递归的申请锁,只要把
attr 的类型改成PTHREAD_MUTEX_RECURSIVE 即可。

1.5、NSConditionLock条件锁

当我们在使用多线程的时候,有时一把只会lock和unlock的锁未必就能完全满足我们的使用。因为普通的锁只能关心锁与不锁,而不在乎用什么钥匙才能开锁,而我们在处理资源共享的时候,多数情况是只有满足一定条件的情况下才能打开这把锁:

  NSMutableArray *products = [NSMutableArray array];
    NSConditionLock *lock = [[NSConditionLock alloc] init];
    NSInteger HAS_DATA = 1;
    NSInteger NO_DATA = 0;

    dispatch_async(dispatch_get_global_queue(0, DISPATCH_QUEUE_PRIORITY_DEFAULT), ^{

        while (1) {
            [lock lockWhenCondition:NO_DATA];
            [products addObject:[[NSObject alloc] init]];
            NSLog(@"produce a product, 总量: %zi", products.count);
            [lock unlockWithCondition:HAS_DATA];
            sleep(1);
        }
    });

    dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{

        while (1) {
            NSLog(@"wait for product");
            [lock lockWhenCondition:HAS_DATA];
            [products removeObjectAtIndex:0];
            NSLog(@"custome a product");
            [lock unlockWithCondition:NO_DATA];
        }
    });

在线程1中的加锁使用了lock,所以是不需要条件的,所以顺利的就锁住了,但在unlock的使用了一个整型的条件,它可以开启其它线程中正在等待这把钥匙的临界地,而线程2则需要一把被标识为2的钥匙,所以当线程1循环到最后一次的时候,才最终打开了线程2中的阻塞。但即便如此,NSConditionLock也跟其它的锁一样,是需要lock与unlock对应的,只是lock,lockWhenCondition:与unlock,unlockWithCondition:是可以随意组合的,当然这是与你的需求相关的。

上面代码执行结果如下:
2017-09-05 16:55:21.740 demo_lock[19324:419225] wait for product
2017-09-05 16:55:21.740 demo_lock[19324:419224] produce a product,
总量: 1
2017-09-05 16:55:21.741 demo_lock[19324:419225] custome a product
2017-09-05 16:55:21.741 demo_lock[19324:419225] wait for product
2017-09-05 16:55:22.743 demo_lock[19324:419224] produce a product,
总量: 1
2017-09-05 16:55:22.743 demo_lock[19324:419225] custome a product
2017-09-05 16:55:22.744 demo_lock[19324:419225] wait for product

互斥锁的实现

互斥锁在申请锁时,调用了 pthread_mutex_lock 方法,它在不同的系统上实现各有不同,有时候它的内部是使用信号量来实现,即使不用信号量,也会调用到 lll_futex_wait 函数,从而导致线程休眠。

上文说到如果临界区很短,忙等的效率也许更高,所以在有些版本的实现中,会首先尝试一定次数(比如
1000 次)的 testandtest,这样可以在错误使用互斥锁时提高性能。

另外,由于 pthread_mutex 有多种类型,可以支持递归锁等,因此在申请加锁时,需要对锁的类型加以判断,这也就是为什么它和信号量的实现类似,但效率略低的原因。

1.6、NSCondition条件锁

 NSCondition *condition = [[NSCondition alloc] init];
    NSMutableArray *products = [NSMutableArray array];

    dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{

        while (1) {
            [condition lock];
            if ([products count] == 0) {
                NSLog(@"wait for product");
                [condition wait];
            }
            [products removeObjectAtIndex:0];
            NSLog(@"custome a product");
            [condition unlock];
        }

    });

    dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{

        while (1) {
            [condition lock];
            [products addObject:[[NSObject alloc] init]];
            NSLog(@"produce a product,总量:%zi",products.count);
            [condition signal];
            [condition unlock];
            sleep(1);
        }

    });

最基本的条件锁。手动控制线程wait和signal

[condition
lock];一般用于多线程同时访问、修改同一个数据源,保证在同一时间内数据源只被访问、修改一次,其他线程的命令需要在lock
外等待,只到unlock ,才可访问

[condition unlock];与lock 同时使用

[condition wait];让当前线程处于等待状态

[condition signal];CPU发信号告诉线程不用在等待,可以继续执行

上面代码执行结果如下:

2017-09-06 11:00:38.925 demo_lock[5145:119574] wait for product
2017-09-06 11:00:42.647 demo_lock[5145:119577] produce a
product,总量:1
2017-09-06 11:00:55.847 demo_lock[5145:119574] custome a product
2017-09-06 11:01:05.752 demo_lock[5145:119574] wait for product
2017-09-06 11:01:08.262 demo_lock[5145:119577] produce a
product,总量:1
2017-09-06 11:01:12.366 demo_lock[5145:119574] custome a product
2017-09-06 11:01:12.366 demo_lock[5145:119574] wait for product

NSLock

NSLock 是 Objective-C
以对象的形式暴露给开发者的一种锁,它的实现非常简单,通过宏,定义了 lock 方法:

#define    MLOCK 
- (void) lock
{
  int err = pthread_mutex_lock(&_mutex);
  // 错误处理 ……
}

NSLock 只是在内部封装了一个 pthread_mutex,属性为 PTHREAD_MUTEX_ERRORCHECK,它会损失一定性能换来错误提示。

这里使用宏定义的原因是,OC 内部还有其他几种锁,他们的 lock
方法都是一模一样,仅仅是内部 pthread_mutex 互斥锁的类型不同。通过宏定义,可以简化方法的定义。

NSLock 比 pthread_mutex 略慢的原因在于它需要经过方法调用,同时由于缓存的存在,多次方法调用不会对性能产生太大的影响。

1.7、pthread_mutex

c 语言定义下多线程枷锁方式

1:pthread_mutex_init(pthread_mutex_t * mutex,const
pthread_mutexattr_t attr);
初始化锁变量mutex。attr为锁属性,NULL值为默认属性。
2:pthread_mutex_lock(pthread_mutex_t
*
mutex);加锁3:pthread_mutex_tylock(**pthread_mutex_t*
*mutex);加锁,但是与2不一样的是当锁已经在使用的时候,返回为EBUSY,而不是挂起等待。4:pthread_mutex_unlock(pthread_mutex_t
*mutex);释放锁5:pthread_mutex_destroy(pthread_mutex_t
*mutex);使用完后释放

   __block pthread_mutex_t theLock;
    pthread_mutex_init(&theLock, NULL);

    dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{

        pthread_mutex_lock(&theLock);
        NSLog(@"需要线程同步的操作1 开始");
        sleep(4);
        NSLog(@"需要线程同步的操作1  结束");
        pthread_mutex_unlock(&theLock);
    });

    dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{

        sleep(1);
        pthread_mutex_lock(&theLock);
        NSLog(@"需要线程同步的操作 2 ");

        pthread_mutex_unlock(&theLock);
    });

代码执行操作结果如下

2017-09-06 14:24:01.867 demo_lock[19902:324468] 需要线程同步的操作1
开始
2017-09-06 14:24:05.871 demo_lock[19902:324468] 需要线程同步的操作1
结束
2017-09-06 14:24:05.872 demo_lock[19902:324469] 需要线程同步的操作 2

NSCondition

NSCondition 的底层是通过条件变量(condition
variable) pthread_cond_t 来实现的。条件变量有点像信号量,提供了线程阻塞与信号机制,因此可以用来阻塞某个线程,并等待某个数据就绪,随后唤醒线程,比如常见的生产者-消费者模式。

1.8、pthread_mutex(recursive)递归锁

    __block pthread_mutex_t theLock;
//    pthread_mutex_init(&theLock, NULL);

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

    dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{

        static void (^ RecursiveMethod)(int);

        RecursiveMethod = ^ (int value) {

            pthread_mutex_lock(&theLock);
            if (value > 0) {
                NSLog(@"value = %d", value);
                sleep(1);
                RecursiveMethod(value - 1);
            }
            pthread_mutex_unlock(&theLock);
        };
        RecursiveMethod(5);
    });

这是pthread_mutex为了防止在递归的情况下出现死锁而出现的递归锁。作用和NSRecursiveLock递归锁类似。

如果使用pthread_mutex_init(&theLock,
NULL);初始化锁的话,上面的代码会出现死锁现象。如果使用递归锁的形式,则没有问题。

打印台输出结果如下

2017-09-06 14:26:54.450 demo_lock[19975:329591] value = 5
2017-09-06 14:26:55.455 demo_lock[19975:329591] value = 4
2017-09-06 14:26:56.456 demo_lock[19975:329591] value = 3
2017-09-06 14:26:57.457 demo_lock[19975:329591] value = 2
2017-09-06 14:26:58.458 demo_lock[19975:329591] value = 1

如何使用条件变量

很多介绍 pthread_cond_t 的文章都会提到,它需要与互斥锁配合使用:

void consumer () { // 消费者  
    pthread_mutex_lock(&mutex);
    while (data == NULL) {
        pthread_cond_wait(&condition_variable_signal, &mutex); // 等待数据
    }
    // --- 有新的数据,以下代码负责处理 ↓↓↓↓↓↓
    // temp = data;
    // --- 有新的数据,以上代码负责处理 ↑↑↑↑↑↑
    pthread_mutex_unlock(&mutex);
}

void producer () {  
    pthread_mutex_lock(&mutex);
    // 生产数据
    pthread_cond_signal(&condition_variable_signal); // 发出信号给消费者,告诉他们有了新的数据
    pthread_mutex_unlock(&mutex);
}

自然我们会有疑问:“如果不用互斥锁,只用条件变量会有什么问题呢?”。问题在于,temp = data; 这段代码不是线程安全的,也许在你把
data
读出来以前,已经有别的线程修改了数据。因此我们需要保证消费者拿到的数据是线程安全的。

wait 方法除了会被 signal 方法唤醒,有时还会被虚假唤醒,所以需要这里
while 循环中的判断来做二次确认。

二、锁的性能对比

在 ibireme 的 不再安全的
OSSpinLock
一文中,有一张图片简单的比较了各种锁的加解锁性能:

澳门新葡亰平台官网 2

性能对比

为什么要使用条件变量

介绍条件变量的文章非常多,但大多都对一个一个基本问题避而不谈:“为什么要用条件变量?它仅仅是控制了线程的执行顺序,用信号量或者互斥锁能不能模拟出类似效果?”

网上的相关资料比较少,我简单说一下个人看法。信号量可以一定程度上替代
condition,但是互斥锁不行。在以上给出的生产者-消费者模式的代码中, pthread_cond_wait 方法的本质是锁的转移,消费者放弃锁,然后生产者获得锁,同理,pthread_cond_signal 则是一个锁从生产者到消费者转移的过程。

如果使用互斥锁,我们需要把代码改成这样:

void consumer () { // 消费者  
    pthread_mutex_lock(&mutex);
    while (data == NULL) {
        pthread_mutex_unlock(&mutex);
        pthread_mutex_lock(&another_lock)  // 相当于 wait 另一个互斥锁
        pthread_mutex_lock(&mutex);
    }
    pthread_mutex_unlock(&mutex);
}

这样做存在的问题在于,在等待 anotherlock 之前,
生产者有可能先执行代码, 从而释放了
another
lock。也就是说,我们无法保证释放锁和等待另一个锁这两个操作是原子性的,也就无法保证“先等待、后释放
another_lock” 这个顺序。

用信号量则不存在这个问题,因为信号量的等待和唤醒并不需要满足先后顺序,信号量只表示有多少个资源可用,因此不存在上述问题。然而与 pthread_cond_wait 保证的原子性锁转移相比,使用信号量似乎存在一定风险(暂时没有查到非原子性操作有何不妥)。

不过,使用 condition
有一个好处,我们可以调用 pthread_cond_broadcast 方法通知所有等待中的消费者,这是使用信号量无法实现的。

三、实现原理

NSCondition 的做法

NSCondition 其实是封装了一个互斥锁和条件变量,
它把前者的 lock 方法和后者的 wait/signal统一在 NSCondition 对象中,暴露给使用者:

- (void) signal {
  pthread_cond_signal(&_condition);
}

// 其实这个函数是通过宏来定义的,展开后就是这样
- (void) lock {
  int err = pthread_mutex_lock(&_mutex);
}

它的加解锁过程与 NSLock 几乎一致,理论上来说耗时也应该一样(实际测试也是如此)。在图中显示它耗时略长,我猜测有可能是测试者在每次加解锁的前后还附带了变量的初始化和销毁操作。

自旋锁的实现原理

自旋锁的目的是为了确保临界区只有一个线程可以访问,伪代码

do {
    Acquire Lock
        Critical section  // 临界区
    Release Lock
        Reminder section // 不需要锁保护的代码
}

自旋锁的实现思路很简单,理论上来说只要定义一个全局变量,用来表示锁的可用情况即可,伪代码如下

bool lock = false; // 一开始没有锁上,任何线程都可以申请锁
do {
    while(lock); // 如果 lock 为 true 就一直死循环,相当于申请锁
    lock = true; // 挂上锁,这样别的线程就无法获得锁
        Critical section  // 临界区
    lock = false; // 相当于释放锁,这样别的线程可以进入临界区
        Reminder section // 不需要锁保护的代码        
}

这段代码存在一个问题: 如果一开始有多个线程同时执行 while
循环,他们都不会在这里卡住,而是继续执行,这样就无法保证锁的可靠性了。解决思路也很简单,只要确保申请锁的过程是原子操作即可。

NSRecursiveLock

上文已经说过,递归锁也是通过 pthread_mutex_lock 函数来实现,在函数内部会判断锁的类型,如果显示是递归锁,就允许递归调用,仅仅将一个计数器加一,锁的释放过程也是同理。

NSRecursiveLock 与 NSLock 的区别在于内部封装的 pthread_mutex_t 对象的类型不同,前者的类型为 PTHREAD_MUTEX_RECURSIVE

原子操作

狭义上:表示一条不可打断的操作,在线程执行操作过程中,不会被操作系统挂起,一定要执行完毕

在多处理器的情况下,能够被多个处理器同时执行的操作不能算原子操作。因此,真正的原子操作必须由硬件提供支持。

我们只需要知道申请锁的过程,可以用一个原子性操作test_and_set来完成
用伪代码这样表示

bool test_and_set (bool *target) {
    bool rv = *target; 
    *target = TRUE; 
    return rv;
}

NSConditionLock

NSConditionLock 借助 NSCondition 来实现,它的本质就是一个生产者-消费者模型。“条件被满足”可以理解为生产者提供了新的内容。NSConditionLock 的内部持有一个 NSCondition 对象,以及_condition_value 属性,在初始化时就会对这个属性进行赋值:

// 简化版代码
- (id) initWithCondition: (NSInteger)value {
    if (nil != (self = [super init])) {
        _condition = [NSCondition new]
        _condition_value = value;
    }
    return self;
}

它的 lockWhenCondition 方法其实就是消费者方法:

- (void) lockWhenCondition: (NSInteger)value {
    [_condition lock];
    while (value != _condition_value) {
        [_condition wait];
    }
}

对应的 unlockWhenCondition 方法则是生产者,使用了 broadcast 方法通知了所有的消费者:

- (void) unlockWithCondition: (NSInteger)value {
    _condition_value = value;
    [_condition broadcast];
    [_condition unlock];
}
自旋锁的总结

原理

bool lock = false; // 一开始没有锁上,任何线程都可以申请锁
do {
    while(test_and_set(&lock); // test_and_set 是一个原子操作
        Critical section  // 临界区
    lock = false; // 相当于释放锁,这样别的线程可以进入临界区
        Reminder section // 不需要锁保护的代码        
}

如果临界区的执行时间过长,使用自旋锁就不是个好主意
;临界区执行时间较长,比如是文件读写,这种忙等是毫无必要的。

@synchronized

这其实是一个 OC 层面的锁, 主要是通过牺牲性能换来语法上的简洁与可读。

我们知道 @synchronized 后面需要紧跟一个 OC
对象,它实际上是把这个对象当做锁来使用。这是通过一个哈希表来实现的,OC
在底层使用了一个互斥锁的数组(你可以理解为锁池),通过对对象去哈希值来得到对应的互斥锁。

具体的实现原理可以参考这篇文章: 关于
@synchronized,这儿比你想知道的还要多

信号量
int sem_wait (sem_t *sem) {
  int *futex = (int *) sem;
  if (atomic_decrement_if_positive (futex) > 0)
    return 0;
  int err = lll_futex_wait (futex, 0);
    return -1;
)

首先会把信号量的值减一,并判断是否大于0,如果大于0,说明不用等待,立刻返回。等待操作在lll_futex_wait函数中实现。使线程进入睡眠状态,主动让出时间片,这个函数在互斥锁的实现中,也有可能被用到。

主动让出时间片并不代表效率高,会导致操作系统切换到另一个线程。上下文切换通常需要10ms,而且需要切换两次。如果等待时间很短,比如只有几微秒,忙等就比线程休眠更高效。

可以看出自旋锁和信号量的实现都非常简单,这也是两者的加解锁时间分别排在第一和第二的原因。

加解锁耗时不能准确反应出锁的效率,只能从一定程度上衡量锁的实现负责程度。

pthread_mutex

phread表示POSIX
thread,定义了一组跨平台的线程相关的API,pthread_mutex表示互斥锁。互斥锁的实现原理与信号量非常相似,不是使用忙等,而是阻塞线程并睡眠,需要进行上下文切换。

pthread_mutexattr_t attr;
pthread_mutexattr_init(&attr);
pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_NORMAL);  // 定义锁的属性

pthread_mutex_t mutex;
pthread_mutex_init(&mutex, &attr) // 创建锁

pthread_mutex_lock(&mutex); // 申请锁
    // 临界区
pthread_mutex_unlock(&mutex); // 释放锁

对于 pthread_mutex
来说,它的用法和之前没有太大的改变,比较重要的是锁的类型,可以有
PTHREAD_MUTEX_NORMALPTHREAD_MUTEX_ERRORCHECKPTHREAD_MUTEX_RECURSIVE
等等,具体的特性就不做解释了,网上有很多相关资料。

一般情况下,一个线程只能申请一次锁,也只能在获得锁的情况下才能释放锁,多次申请锁或释放未获得的锁都会导致崩溃。假设在已经获得锁的情况下再次申请锁,线程会因为等待锁的释放而进入睡眠状态,因此就不可能再释放锁,从而导致死锁。

然而这种情况经常会发生,比如某个函数申请了锁,在临界区内又递归调用了自己。辛运的是
pthread_mutex 支持递归锁,也就是允许一个线程递归的申请锁,只要把 attr
的类型改成 PTHREAD_MUTEX_RECURSIVE 即可。

互斥锁的实现

互斥锁在申请锁时,调用了 pthread_mutex_lock
方法,它在不同的系统上实现各有不同,有时候它的内部是使用信号量来实现,即使不用信号量,也会调用到
lll_futex_wait 函数,从而导致线程休眠。

上文说到如果临界区很短,忙等的效率也许更高,所以在有些版本的实现中,会首先尝试一定次数(比如
1000 次)的 test_and_test,这样可以在错误使用互斥锁时提高性能。

另外,由于 pthread_mutex
有多种类型,可以支持递归锁等,因此在申请加锁时,需要对锁的类型加以判断,这也就是为什么它和信号量的实现类似,但效率略低的原因。

NSLock

NSLock是Objective-C
已对象的形式暴露给开发者的一种锁,实现简单,通过宏,定义了lock方法:

#define    MLOCK 
- (void) lock
{
  int err = pthread_mutex_lock(&_mutex);
  // 错误处理 ……
}

NSLock在内部封装了phread_mutex,属性类型为PTHREAD_MUTEX_ERRORCHECK,它会损失一定性能换来错误提示。

NSLockpthread_mutex略慢的原因是在于它需要经过方法调用,但是由于缓存的存在,多次方法调用不会对性能产生太大影响。

NSCondition

NSCondititon的底层是通过条件变量(condition
variable)pthread_cond_t来实现的。条件变量有些像信号量,提供了线程阻塞与信号机制,因此可以用来阻塞某个线程,并等待某个数据就绪,随后唤醒线程,比如常见的生产者-消费模式

很多介绍pthread_cond_t的文章都会提到,它需要与互斥锁配合使用

void consumer () { // 消费者
    pthread_mutex_lock(&mutex);
    while (data == NULL) {
        pthread_cond_wait(&condition_variable_signal, &mutex); // 等待数据
    }
    // --- 有新的数据,以下代码负责处理 ↓↓↓↓↓↓
    // temp = data;
    // --- 有新的数据,以上代码负责处理 ↑↑↑↑↑↑
    pthread_mutex_unlock(&mutex);
}

void producer () {
    pthread_mutex_lock(&mutex);
    // 生产数据
    pthread_cond_signal(&condition_variable_signal); // 发出信号给消费者,告诉他们有了新的数据
    pthread_mutex_unlock(&mutex);
}
NSCondition的做法

NSCondition其实是封装了一个互斥锁和条件变量,把前者的lock方法和后者的wait/signal统一在NSCondition对象中,暴露给使用者:

- (void) signal {
  pthread_cond_signal(&_condition);
}

// 其实这个函数是通过宏来定义的,展开后就是这样
- (void) lock {
  int err = pthread_mutex_lock(&_mutex);
}
NSRecursiveLock

上文已经说过,递归锁也是通过 pthread_mutex_lock
函数来实现,在函数内部会判断锁的类型,如果显示是递归锁,就允许递归调用,仅仅将一个计数器加一,锁的释放过程也是同理。

NSRecursiveLockNSLock 的区别在于内部封装的 pthread_mutex_t
对象的类型不同,前者的类型为 PTHREAD_MUTEX_RECURSIVE

NSConditionLock

NSConditionLock借助NSCondition来实现,它的本质就是一个生产者-消费者模型。“条件被满足”可以理解为生产者提供了新的内容。NSConditionLock
的内部持有一个 NSCondition 对象,以及 _condition_value
属性,在初始化时就会对这个属性进行赋值:

- (id) initWithCondition: (NSInteger)value {
    if (nil != (self = [super init])) {
        _condition = [NSCondition new]
        _condition_value = value;
    }
    return self;
}

它的lockWhenCondition方法其实就是消费者方法

- (void) lockWhenCondition: (NSInteger)value {
    [_condition lock];
    while (value != _condition_value) {
        [_condition wait];
    }
}

对应的unlockWhenCondition方法则是生产者,使用了broadcast方法通知了所有的消费者

- (void) unlockWithCondition: (NSInteger)value {
    _condition_value = value;
    [_condition broadcast];
    [_condition unlock];
}
@synchronized

这其实是一个 OC 层面的锁, 主要是通过牺牲性能换来语法上的简洁与可读。

我们知道 @synchronized 后面需要紧跟一个 OC
对象,它实际上是把这个对象当做锁来使用。这是通过一个哈希表来实现的,OC
在底层使用了一个互斥锁的数组(你可以理解为锁池),通过对对象去哈希值来得到对应的互斥锁。

参考文献

@synchronized原理
深入理解iOS开发中的锁
不再安全的
OSSpinLock

You can leave a response, or trackback from your own site.

Leave a Reply

网站地图xml地图