Python3 与 C# 并发编程之～线程篇3

原创

逸鹏

发布于 2018-09-07 00:39:22

48400

代码可运行

文章被收录于专栏：逸鹏说道逸鹏说道

运行总次数：0

代码可运行

锁专题扩展

1.加锁机制

在多线程程序中，死锁问题很大一部分是由于线程同时获取多个锁造成的，eg：一个线程获取了第一个锁，然后在获取第二个锁的时候发生阻塞，那么这个线程就可能阻塞其他线程的执行，从而导致整个程序假死。

解决死锁问题的一种方案是为程序中的每一个锁分配一个唯一的id，然后只允许按照升序规则来使用多个锁，当时举了个小明小张转账的简单例子，来避免死锁，这次咱们再看一个案例：（这个规则使用上下文管理器非常简单）

先看看源码，咱们怎么使用：

# 装饰器方法def contextmanager(func):    """    方法格式    @contextmanager    def some_generator(<arguments>):        <setup>        try:            yield <value>        finally:            <cleanup>    然后就可以直接使用with托管了    with some_generator(<arguments>) as <variable>:        <body>    """    @wraps(func)    def helper(*args, **kwds):        return _GeneratorContextManager(func, args, kwds)    return helper

翻译成代码就是这样了：（简化）

from contextlib import contextmanager  # 引入上下文管理器@contextmanagerdef lock_manager(*args):    # 先排个序（按照id排序）    args = sorted(args, key=lambda x: id(x))    try:        for lock in args:            lock.acquire()        yield    finally:        # 先释放最后加的锁（倒序释放）        for lock in reversed(args):            lock.release()

基础忘记了可以点我（lambda）

以上面小明小张转账案例为例子：（不用再管锁顺序之类的了，直接全部丢进去： withlock_manager(...)）

from contextlib import contextmanager  # 引入上下文管理器from multiprocessing.dummy import Pool as ThreadPool, Lock@contextmanagerdef lock_manager(*args):    # 先排个序（按照id排序）    args = sorted(args, key=lambda x: id(x))    try:        for lock in args:            lock.acquire()        yield    finally:        # 先释放最后加的锁（倒序释放）        for lock in reversed(args):            lock.release()xiaoming = 5000xiaozhang = 8000m_lock = Lock()  # 小明的锁z_lock = Lock()  # 小张的锁# 小明转账1000给小张def a_to_b():    global xiaoming    global xiaozhang    global m_lock    global z_lock    print(f"[转账前]小明{xiaoming},小张{xiaozhang}")    with lock_manager(m_lock, z_lock):        xiaoming -= 1000        xiaozhang += 1000    print(f"[转账后]小明{xiaoming},小张{xiaozhang}")# 小张转账1000给小明def b_to_a():    global xiaoming    global xiaozhang    global m_lock    global z_lock    print(f"[转账前]小明{xiaoming},小张{xiaozhang}")    with lock_manager(m_lock, z_lock):        xiaozhang -= 1000        xiaoming += 1000    print(f"[转账后]小明{xiaoming},小张{xiaozhang}")def main():    print(f"[互刷之前]小明{xiaoming},小张{xiaozhang}")    p = ThreadPool()    for _ in range(5):        p.apply_async(a_to_b)        p.apply_async(b_to_a)    p.close()    p.join()    print(f"[互刷之后]小明{xiaoming},小张{xiaozhang}")if __name__ == '__main__':    main()

输出：

[互刷之前]小明5000,小张8000[转账前]小明5000,小张8000[转账前]小明5000,小张8000[转账后]小明4000,小张9000[转账前]小明5000,小张8000[转账后]小明5000,小张8000[转账前]小明5000,小张8000[转账前]小明4000,小张9000[转账后]小明4000,小张9000[转账后]小明5000,小张8000[转账前]小明5000,小张8000[转账后]小明4000,小张9000[转账前]小明4000,小张9000[转账前]小明4000,小张9000[转账后]小明5000,小张8000[转账前]小明5000,小张8000[转账后]小明4000,小张9000[转账后]小明5000,小张8000[转账前]小明5000,小张8000[转账后]小明4000,小张9000[转账后]小明5000,小张8000[互刷之后]小明5000,小张8000

再来个验证，在他们互刷的过程中，小潘还了1000元给小明

from time import sleepfrom contextlib import contextmanager  # 引入上下文管理器from multiprocessing.dummy import Pool as ThreadPool, Lock@contextmanagerdef lock_manager(*args):    # 先排个序（按照id排序）    args = sorted(args, key=lambda x: id(x))    try:        for lock in args:            lock.acquire()        yield    finally:        # 先释放最后加的锁（倒序释放）        for lock in reversed(args):            lock.release()xiaopan = 9000xiaoming = 5000xiaozhang = 8000m_lock = Lock()  # 小明的锁z_lock = Lock()  # 小张的锁p_lock = Lock()  # 小潘的锁# 小明转账1000给小张def a_to_b():    global xiaoming    global xiaozhang    global m_lock    global z_lock    print(f"[转账前]小明{xiaoming},小张{xiaozhang}")    with lock_manager(m_lock, z_lock):        xiaoming -= 1000        xiaozhang += 1000    print(f"[转账后]小明{xiaoming},小张{xiaozhang}")# 小张转账1000给小明def b_to_a():    global xiaoming    global xiaozhang    global m_lock    global z_lock    print(f"[转账前]小明{xiaoming},小张{xiaozhang}")    with lock_manager(m_lock, z_lock):        xiaozhang -= 1000        xiaoming += 1000    print(f"[转账后]小明{xiaoming},小张{xiaozhang}")# 小潘还1000给小明def c_to_a():    global xiaoming    global xiaopan    global m_lock    global p_lock    print(f"[转账前]小明{xiaoming},小潘{xiaopan}")    with lock_manager(m_lock, p_lock):        xiaopan -= 1000        xiaoming += 1000    print(f"[转账后]小明{xiaoming},小潘{xiaopan}")def main():    print(f"[互刷之前]小明{xiaoming},小张{xiaozhang},小潘{xiaopan}")    p = ThreadPool()    for _ in range(5):        p.apply_async(a_to_b)        # 在他们互刷的过程中，小潘还了1000元给小明        if _ == 3:            p.apply_async(c_to_a)        p.apply_async(b_to_a)    p.close()    p.join()    print(f"[互刷之后]小明{xiaoming},小张{xiaozhang},小潘{xiaopan}")if __name__ == '__main__':    main()

输出：

[互刷之前]小明5000,小张8000,小潘9000[转账前]小明5000,小张8000[转账前]小明5000,小张8000[转账后]小明4000,小张9000[转账前]小明5000,小张8000[转账前]小明4000,小张9000[转账后]小明5000,小张8000[转账前]小明5000,小张8000[转账后]小明4000,小张9000[转账后]小明5000,小张8000[转账前]小明5000,小张8000[转账后]小明4000,小张9000[转账前]小明4000,小张9000[转账前]小明4000,小潘9000 # 注意下这个[转账后]小明5000,小张8000[转账前]小明5000,小张8000[转账后]小明4000,小张9000[转账后]小明5000,小潘8000 # 注意下这个[转账前]小明5000,小张9000[转账后]小明6000,小张8000[转账后]小明5000,小张9000[转账前]小明6000,小张8000[转账后]小明6000,小张8000[互刷之后]小明6000,小张8000,小潘8000

上下文管理器进一步完善

from contextlib import contextmanagerfrom multiprocessing.dummy import threading # or import threading# ThreadLocal 下节会说_local = threading.local()@contextmanagerdef acquire(*args):    # 以id将锁进行排序    args = sorted(args, key=lambda x: id(x))    # 确保不违反以前获取的锁顺序    acquired = getattr(_local, 'acquired', [])    if acquired and max(id(lock) for lock in acquired) >= id(args[0]):        raise RuntimeError('锁顺序有问题')    # 获取所有锁    acquired.extend(args)    _local.acquired = acquired  # ThreadLocal：每个线程独享acquired    # 固定格式    try:        for lock in args:            lock.acquire()        yield    finally:        # 逆向释放锁资源        for lock in reversed(args):            lock.release()        # 把释放掉的锁给删了        del acquired[-len(args):]

2.哲学家吃面

先看看场景：五个外国哲学家到中国来吃饭了，因为不了解行情，每个人只拿了一双筷子，然后点了一大份的面。碍于面子，他们不想再去拿筷子了，于是就想通过脑子来解决这个问题。

每个哲学家吃面都是需要两只筷子的，这样问题就来了：（只能拿自己两手边的筷子）

如果大家都是先拿自己筷子，再去抢别人的筷子，那么就都等着饿死了（死锁）
如果有一个人打破这个常规，先拿别人的筷子再拿自己的，那么肯定有一个人可以吃到面了
5个筷子，意味着最好的情况 ==> 同一时刻有2人在吃（0人，1人，2人）

把现实问题转换成代码就是：

哲学家--线程
筷子--资源（几个资源对应几把锁）
吃完一口面就放下筷子--lock的释放

有了上面基础这个就简单了，使用死锁避免机制解决哲学家就餐问题的实现：（不用再操心锁顺序了）

from contextlib import contextmanager  # 引入上下文管理器from multiprocessing.dummy import Pool as ThreadPool, Lock, current_process as current_thread# 使用简化版，便于你们理解@contextmanagerdef lock_manager(*args):    # 先排个序（按照id排序）    args = sorted(args, key=lambda x: id(x))    try:        # 依次加锁        for lock in args:            lock.acquire()        yield    finally:        # 先释放最后加的锁（倒序释放）        for lock in reversed(args):            lock.release()#########################################def eat(l_lock, r_lock):    while True:        with lock_manager(l_lock, r_lock):            # 获取当前线程的名字            print(f"{current_thread().name}，正在吃面")            sleep(0.5)def main():    resource = 5  # 5个筷子，5个哲学家    locks = [Lock() for i in range(resource)]  # 几个资源几个锁    p = ThreadPool(resource) # 让线程池里面有5个线程（默认是cup核数）    for i in range(resource):        # 抢左手筷子（locks[i]）和右手的筷子（locks[(i + 1) % resource]）        # 举个例子更清楚：i=0 ==> 0,1；i=4 ==> 4,0        p.apply_async(eat, args=(locks[i], locks[(i + 1) % resource]))    p.close()    p.join()if __name__ == '__main__':    main()

输出图示：

自行拓展

1.银行家算法

PS：这个一般都是操作系统的算法，了解下就可以了，上面哲学家吃面用的更多一点（欢迎投稿～）

我们可以把操作系统看作是银行家，操作系统管理的资源相当于银行家管理的资金，进程向操作系统请求分配资源相当于用户向银行家贷款。为保证资金的安全，银行家规定：

当一个顾客对资金的最大需求量不超过银行家现有的资金时就可接纳该顾客；
顾客可以分期贷款，但贷款的总数不能超过最大需求量；
当银行家现有的资金不能满足顾客尚需的贷款数额时，对顾客的贷款可推迟支付，但总能使顾客在有限的时间里得到贷款；
当顾客得到所需的全部资金后，一定能在有限的时间里归还所有的资金.

操作系统按照银行家制定的规则为进程分配资源，当进程首次申请资源时，要测试该进程对资源的最大需求量，如果系统现存的资源可以满足它的最大需求量则按当前的申请量分配资源，否则就推迟分配。当进程在执行中继续申请资源时，先测试该进程本次申请的资源数是否超过了该资源所剩余的总量。若超过则拒绝分配资源，若能满足则按当前的申请量分配资源，否则也要推迟分配。

通俗讲就是：当一个进程申请使用资源的时候，银行家算法通过先试探分配给该进程资源，然后通过安全性算法判断分配后的系统是否处于安全状态，若不安全则试探分配作废，让该进程继续等待。

参考链接：

https://www.cnblogs.com/chuxiuhong/p/6103928.htmlhttps://www.cnblogs.com/Lynn-Zhang/p/5672080.htmlhttps://blog.csdn.net/qq_33414271/article/details/80245715https://blog.csdn.net/qq_37315403/article/details/82179707