【操作系统知识】线程间通信

落日映苍穹つ 2023-02-24 11:24 19阅读 0赞

## 9. 线程间通信 ##

### (1) 互斥锁 ###

> 互斥锁为资源引入一个状态：锁定/非锁定。某个线程要更改共享数据时，先将其锁定，此时资源的状态为“锁定”，其他线程不能更改；直到该线程释放资源，将资源的状态变成“非锁定”，其他的线程才能再次锁定该资源。互斥锁保证了每次只有一个线程进行写入操作，从而保证了多线程情况下数据的正确性。

"""
    #创建锁
    mutex = threading.Lock()
    #锁定
    mutex.acquire([timeout])
    #释放
    mutex.release()
    """
    import threading
    
    money = 0
    
    def order(n):
        global money
        money = money + n
        money = money - n
    
    class MyThread(threading.Thread):
        def __init__(self, thread_name):
            threading.Thread.__init__(self, name='线程' + thread_name)
            self.thread_name = int(thread_name)
    
        def run(self):
            for i in range(1000000):
                lock.acquire()  # 加锁
                order(self.thread_name)
                lock.release()  # 释放
    
    if __name__ == '__main__':
        lock = threading.Lock()
        t1 = MyThread('1')
        t2 = MyThread('10')
        t1.start()
        t2.start()
        t1.join()
        t2.join()
        print(money)
    
    # 结果
    '''
    0
    '''

> 为了支持在同一线程中多次请求同一资源，python提供了“可重入锁”：threading.RLock。RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次acquire。直到一个线程所有的acquire都被release，其他的线程才能获得资源。acquire() 和 release() 必须成对出现，也就是说加了几把锁就得释放几把锁。

import threading
    import time
    
    num = 0
    mutex = threading.RLock()
    
    class MyThread(threading.Thread):
        def run(self):
            global num
            time.sleep(1)
    
            if mutex.acquire():
                num = num + 1
                msg = self.name + 'set num to' + str(num)
                print(msg)
                mutex.acquire()
                mutex.release()
                mutex.release()
    
    if __name__ == '__main__':
        for i in range(5):
            t = MyThread()
            t.start()
    
    # 结果
    '''
    Thread-1set num to1
    Thread-4set num to2
    Thread-3set num to3
    Thread-2set num to4
    Thread-5set num to5
    '''

### (2) 信号量 ###

> 在多线程编程中，为了防止不同的线程同时对一个公用的资源（比如全部变量）进行修改，需要进行同时访问的数量（通常是1）的限制。信号量同步基于内部计数器，每调用一次acquire()，计数器减1；每调用一次release()，计数器加1.当计数器为0时，acquire()调用被阻塞。

from threading import Semaphore, Thread
    import time
    
    # 创建信号量
    sema = Semaphore(3)  # 限制同时访问资源的数量为3
    
    
    def foo(tid):
        with sema:
            print(f"{tid} acquire sema")
            time.sleep(1)
        print(f"{tid} release sema")
    
    
    threads = []
    
    for i in range(5):
        t = Thread(target=foo, args=(i,))
        threads.append(t)
        t.start()
    
    for t in threads:
        t.join()
    
    # 结果
    '''
    0 acquire sema
    1 acquire sema
    2 acquire sema
    1 release sema3 acquire sema
    0 release sema
    
    4 acquire sema
    2 release sema
    3 release sema
    4 release sema
    
    '''

### (3) 条件变量 ###

> 条件变量可以认为是更高级的锁，比Lock和Rlock的用法更高级，能处理一些复杂的线程同步问题。threading.Condition() 创建一把资源锁（默认是Rlock），提供 acquire() 和 release() 方法，用法和 Rlock 一致。此外 Condition 还提供 wait()、Notify() 和 NotifyAll() 方法。

> wait()：线程挂起，直到收到一个 Notify() 通知或者超时（可选参数），wait() 必须在线程得到 Rlock 后才能使用。

> Notify() ：在线程挂起的时候，发送一个通知，让 wait() 等待线程继续运行，Notify() 也必须在线程得到 Rlock 后才能使用。 Notify(n=1)，最多唤醒 n 个线程。

> NotifyAll() ：在线程挂起的时候，发送通知，让所有 wait() 阻塞的线程都继续运行。

import threading
    import time
    
    
    def test_a():
        condition.acquire()
        print("李白：中路团一波，推塔")
        condition.wait()
        print("李白：好的")
        condition.notify()
        condition.release()
    
    
    def test_b():
        time.sleep(2)
        condition.acquire()
        print("武则天：等我大冷却")
        condition.notify()
        condition.wait()
        print("武则天：我的大好了，干干干")
    
    
    if __name__ == '__main__':
        condition = threading.Condition()
        ta = threading.Thread(target=test_a)
        tb = threading.Thread(target=test_b)
        ta.start()
        tb.start()
        ta.join()
        tb.join()
    
    # 结果
    '''
    李白：中路团一波，推塔
    武则天：等我大冷却
    李白：好的
    武则天：我的大好了，干干干
    '''

### (4) 事件 ###

> 线程的一个关键特性是每个线程都是独立运行且状态不可预测。如果程序中的其他线程需要通过断某个线程的状态来确定自己下一步的操作，这时线程同步问题就会变得非常棘手。为了解决这些问题，我们需要使用 threading 库中的 Event 对象。Event 对象包含一个可由线程设置的信号标志，它允许线程等待某些事件的发生。在初始情况下，event 对象中的信号标志被设置假。如果有线程等待一个 event 对象，而这个 event 对象的标志为假，那么这个线程将会被一直阻塞，直至该标志为真。一个线程如果将一个 event 对象的信号标志设置为真，它将唤醒所有等待这个 event 对象的线程。如果一个线程等待一个已经被设置为真的 event 对象，那么它将忽略这个事件，继续执行。

"""
    event.wait(): 阻塞线程，知道flag值变为True
    event.set(): 设置flag值为True
    event.clear(): 修改flag值为False
    event.isSet()：仅当flag值为True时返回
    """
    # 主线程启动子线程后sleep两秒，子线程因为event.wait()被阻塞，当主线程醒来后执行event.set()，子线程才继续运行，两者输出时间差2s
    
    import threading
    import time
    import datetime
    
    class thread(threading.Thread):
        def __init__(self, thread_name):
            threading.Thread.__init__(self, name='线程' + thread_name)
            self.thread_name = int(thread_name)
    
        def run(self):
            event.wait()
            print(f"子线程运行时间: {datetime.datetime.now()}")
    
    if __name__ == '__main__':
        event = threading.Event()
        t1 = thread('0')
        t1.start()  # 启动子线程
        print(f"主线程运行时间：{datetime.datetime.now()}")
        time.sleep(2)
        event.set()  # flag设置为True
        t1.join()
    
    # 结果
    '''
    主线程运行时间：2020-06-15 21:44:54.221804
    子线程运行时间: 2020-06-15 21:44:56.222719
    '''

### (5) 消息队列 ###

> 使用线程队列有一个要注意的问题是，向队列中添加数据项时并不会复制此数据项，线程间通信实际上是在线程间传递对象引用。如果你担心对象的共享状态，那你最好只传递不可修改的数据结构（如：整型、字符串或者元组）或者一个对象的深拷贝。

> 在一个进程中，不同子线程负责不同的任务，t1子线程负责获取到数据，t2子线程负责把数据保存的本地，那么他们之间的通信使用Queue来完成。因为在一个进程中，数据变量是共享的，即多个子线程可以对同一个全局变量进行操作修改，Queue是加了锁的安全消息队列。

# 假设需要爬取博客网站的所有文章详情，先要通过文章列表页爬取所有文章的url，再根据文章的url，爬取文章的具体内容
    import threading
    from queue import Queue
    import time
    
    # 爬取文章详情页
    def get_detail_html(detail_url_list, id):
        while True:
            url = detail_url_list.get()
            time.sleep(2)  # 延时2s，模拟网络请求
            print("thread {id}: get {url} detail finished ".format(id=id,url=url))
    
    # 爬取文章列表页
    def get_detail_url(queue):
        for i in range(100):
            time.sleep(1) # 延时1s，模拟比爬取文章详情要快
            queue.put("http://projectedu.com/{id}".format(id=i))
            print("get detail url {id} end".format(id=i))
    
    if __name__ == "__main__":
        detail_url_queue = Queue(maxsize=1000)
        # 先创造两个线程
        thread = threading.Thread(target=get_detail_url, args=(detail_url_queue,))
        html_thread= []
        for i in range(3):
            thread2 = threading.Thread(target=get_detail_html, args=(detail_url_queue,i))
            html_thread.append(thread2)
        start_time = time.time()
        # 启动两个线程
        thread.start()
        for i in range(3):
            html_thread[i].start()
        # 等待所有线程结束
        thread.join()
        for i in range(3):
            html_thread[i].join()
    
        print("last time: {} s".format(time.time()-start_time))
    
    # 结果
    '''
    get detail url 0 end
    get detail url 1 end
    thread 0: get http://projectedu.com/0 detail finished 
    get detail url 2 end
    thread 1: get http://projectedu.com/1 detail finished 
    get detail url 3 end
    thread 2: get http://projectedu.com/2 detail finished 
    get detail url 4 end
    thread 0: get http://projectedu.com/3 detail finished 
    get detail url 5 end
    thread 1: get http://projectedu.com/4 detail finished 
    get detail url 6 end
    thread 2: get http://projectedu.com/5 detail finished 
    get detail url 7 end
    '''