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()