python3.6 - threading 多线程编程进阶，线程间并发控制（2）

󰃭 2017-07-25

1. 线程间同步简介

Condition 条件变量
Semaphore 信号量
Barrier 栅栏/屏障
Timer 定时器

2. Condition 条件变量

定义

条件变量变量是来自 POSIX 线程标准的一种线程同步机制。主要用来等待某个条件的发生，用来同步同一进程中的多个线程。条件变量总是和一个互斥量或锁关联，条件本身由互斥量保护，线程在改变条件状态时必须锁住互斥量。线程可以等待在某个条件上，不需要通过轮询的方式来判断，节省CPU时间

使用方法

条件变量需要配合着锁（或者互斥量）一起使用，条件本身接受锁的保护，控制线程的资源访问
条件变量使用 with 语法自动调用关联锁的获取（acquire），释放（release）方法
wait 方法会释放锁，并且阻塞当前线程，直到被其他线程的 notify/notify_all 唤醒，一旦阻塞的线程被从 wait 唤醒，将重新获取锁
用 notify 方法唤醒等待条件变量的一个线程，notify_all 方法唤醒等待条件变量的所有线程
notify 和 notify_all 方法不会释放锁，也就是说被唤醒的线程不会立即从 wait 方法返回，需要调用 notify 和 notify_all 的线程交出锁的所有权才能返回

使用示例

生产者，消费者模型，使用条件变量控制并发访问普通的 list

#!/usr/bin/env python
# coding: utf-8


import time
import threading
import random
import itertools


# 全局变量
# 停止标志
is_stop = False

# 队列最大值
MAX_NUM = 100

# 队列对象
data_queue = []


class Producer(threading.Thread):
    '''
    生产者线程对象
    name: 线程名字
    condition: 条件变量
    '''
    def __init__(self, name, condition):
        self._oname = name
        self._condition = condition
        super().__init__()

    def run(self):
        '''
        生产过程
        '''
        while True:
            with self._condition:
                while len(data_queue) >= MAX_NUM:
                    print('Producer oname={}, to wait'.format(self._oname))
                    self._condition.wait()

                # 判断是否结束生产
                if is_stop:
                    return

                data_queue.append('Producer oname={}, value={}'
                                  .format(self._oname, random.random()))
                info = 'Producer oname={}, queue_size={}' \
                        .format(self._oname, len(data_queue))
                print(info)
                self._condition.notify()
            time.sleep(random.random())


class Consumer(threading.Thread):
    '''
    消费者线程对象
    name: 线程名字
    condition: 条件变量
    '''
    def __init__(self, name, condition):
        self._oname = name
        self._condition = condition
        super().__init__()

    def run(self):
        '''
        消费过程
        '''
        while True:
            with self._condition:
                while not data_queue:
                    print('    Consumer oname={}, to wait'
                          .format(self._oname))

                    # 判断是否结束消费
                    if is_stop:
                        return

                    self._condition.wait()
                data = data_queue.pop(0)
                size = len(data_queue)
                self._condition.notify()

            print('    Consumer oname={}, data={}, queue_size={}'
                  .format(self._oname, data, size))
            time.sleep(random.random())


def main():
    '''
    验证函数
    '''
    # 创建条件变量，供生产者线程，消费者线程使用
    condition = threading.Condition()

    # 创建生产者线程组
    producerts_thr = [Producer('Producer_{}'.format(i), condition)
                      for i in range(4)]
    # 创建消费者线程组
    consumers_thr = [Consumer('Consumer_{}'.format(i), condition)
                     for i in range(4)]

    # 开始生产者，消费者线程
    [i.start() for i in itertools.chain(producerts_thr, consumers_thr)]

    # 定义停止计时器函数
    def stop_all():
        with condition:
            global is_stop
            is_stop = True

    # 设定运行 10 秒钟之后停止生产者，消费者线程
    t = threading.Timer(10.0, stop_all)
    t.start()

    # 等待生产者，消费者线程结束
    [i.join() for i in itertools.chain(producerts_thr, consumers_thr)]


if __name__ == '__main__':
    main()

使用方法

acquire 减少计数，计数器是 0 的时候阻塞
release 增加计数

使用示例

#!/usr/bin/env python
# coding: utf-8


import threading
import random
import time


def work_thr(pool_sema, num):
    '''
    线程函数
    '''
    while True:
        # 每次只允许固定数目的线程进入
        with pool_sema:
            print('thr num={}'.format(num))

        time.sleep(3)


if __name__ == '__main__':
    # 创建允许 5 个线程访问的 semaphore
    pool_sema = threading.BoundedSemaphore(5)

    # 创建 20 个线程
    thrs = [threading.Thread(target=work_thr, args=[pool_sema, i])
            for i in range(20)]

    # 开始所有线程
    [thr.start() for thr in thrs]

    # 等待线程结束
    [thr.join() for thr in thrs]

使用示例（伪代码）

同步客户端和服务器的例子

#!/usr/bin/env python
# coding: utf-8


import time
from threading import Barrier, Thread


b = Barrier(2, timeout=5)

def server():
    # start_server()
    time.sleep(3)

    # 服务端到达
    b.wait()
    print('start server')

def client():
    time.sleep(5)
    # 客户端到达
    b.wait()
    print('start client')

if __name__ == '__main__':
    # 创建线程
    serv_thr = Thread(target=server)
    cli_thr = Thread(target=client)

    # 开始执行线程
    serv_thr.start()
    cli_thr.start()

    # 等待线程退出
    serv_thr.join()
    cli_thr.join()

使用示例

def hello():
    print("hello, world")

# 30 秒后打印 "hello, world"
t = Timer(30.0, hello)
t.start()