Python并发之多线程threading(1)

5, 使用事件Event控制线程同步

Event是线程通信最为简单的机制，一个线程抛出一个信号，另外线程等待这个信号。

事件是线程之间用于通讯的对象。有的线程等待信号，有的线程发出信号。基本上事件对象都会维护一个内部变量，可以通过 set() 方法设置为 true ，也可以通过 clear() 方法设置为 false。 wait() 方法将会阻塞线程，直到内部变量为 true 。

Event（事件）处理的机制：全局定义了一个内置标志Flag，如果Flag值为 False，那么当程序执行 event.wait方法时就会阻塞，如果Flag值为True，那么event.wait 方法时便不再阻塞。

Event其实就是一个简化版的 Condition。Event没有锁，无法使线程进入同步阻塞状态。当多个线程监听某一个触发条件时，如果全部线程都依赖这个触发条件，可以使用Event；如果需要精确控制某些线程被触发，则需要使用Condition。

Event()方法如下：

示例：交通灯控制车辆，绿灯行红灯停

from threading import Thread, Eventimport timeclass Trafficlight(Thread):    def __init__(self, event, green_cnt, red_cnt):        super(Trafficlight, self).__init__()        self.event = event        self.green_cnt = green_cnt        self.red_cnt = red_cnt        self.interval = self.green_cnt + self.red_cnt    def run(self):        count = 0        self.event.set()   ##绿灯亮        while True:            if count <= self.green_cnt:                self.event.set() ##绿灯亮                print("绿灯亮了 {}s，请通行".format(count))            elif self.green_cnt < count <= self.interval:                self.event.clear() ##红灯亮                print("红灯亮 {}s，车辆禁止通行".format(count))            elif count > self.interval:                self.event.set() ##绿灯亮                count = 0                print("绿灯亮了 {}s，请通行".format(count))            time.sleep(1)            count = count + 1class Car(Thread):    def __init__(self, event, car_name):        super(Car, self).__init__()        self.event = event        self.car_name = car_name    def run(self):        while True:            if self.event.is_set(): # 有标志位，代表是绿灯                print("{} is runnning".format(self.car_name))                time.sleep(1)            else: # 如果不是绿灯就代表红灯                self.event.wait() ##阻塞，直到event被set才继续执行                print("{} is waiting".format(self.car_name))if __name__ == "__main__":    event = Event()    thread_List = []    car_List = ["BMW", "Ford", "BenZ"]    thread_light = Trafficlight(event, 1, 2)    thread_List.append(thread_light)    for car_name in car_List:        car_th = Car(event, car_name) ##所有的车收到event事件同时开始执行/阻塞        thread_List.append(car_th)    for th in thread_List:        th.start()    for th in thread_List:        th.join()##输出如下：绿灯亮了 0s，请通行BMW is runnningFord is runnningBenZ is runnningBMW is runnningFord is runnning绿灯亮了 1s，请通行BenZ is runnningBMW is runnning红灯亮 2s，车辆禁止通行红灯亮 3s，车辆禁止通行

import threadingfrom threading import barrIEr, Threaddef prepare_user_data(user):    print("user {} is ready".format(user))def prepare_car_data(car):    print("car {} is ready".format(car))def ready():    print("{} 数据准备好了".format(threading.current_thread().name))class RunThreading(Thread):    def __init__(self, barrIEs, target=None, args=(), kwargs={}):        super(RunThreading, self).__init__(target=None, args=args, kwargs=kwargs)        self._barrIEs = barrIEs        self.func = target        self.args = args        self._kwargs = kwargs    def run(self):        print("data prepare......")        data = self.func(*self.args, **self._kwargs)        self._barrIEs.wait()        return dataif __name__ == "__main__":    data = {"Leo": "沪111111", "Bruce": "沪222222", "Frank": "沪333333"}    cnt = len(data)    print("并发测试数据准备...")    barrIEr = barrIEr(cnt, action=ready, timeout=10)    thread_List = []    for k, v in data.items():        thread_user = RunThreading(barrIEr, target=prepare_user_data, args=(k,))        thread_car = RunThreading(barrIEr, target=prepare_car_data, args=(v,))        thread_List.append(thread_user)        thread_List.append(thread_car)    for thread in thread_List:        thread.start()    for thread in thread_List:        thread.join()    if barrIEr.broken:        print("数据准备出异常了，请重新准备数据")

并发测试数据准备...data prepare......user Leo is readydata prepare......car 沪111111 is readydata prepare......user Bruce is readyThread-3 数据准备好了data prepare......car 沪222222 is readydata prepare......user Frank is readydata prepare......car 沪333333 is readyThread-6 数据准备好了

from threading import Threadfrom queue import Queueimport random, timeclass Producer(Thread):    def __init__(self, queue):        super(Producer, self).__init__()        self.queue = queue    def run(self):        for i in range(10):            item = random.randint(0, 256)            self.queue.put(item) ##插入队列            print('Producer notify: item N° %d appended to queue by %s' % (item, self.name))            time.sleep(1)class Consumer(Thread):    def __init__(self, queue):        Thread.__init__(self)        self.queue = queue    def run(self):        while True:            item = self.queue.get() ##从队列中取出数据            print('Consumer notify : %d popped from queue by %s' % (item, self.name))            self.queue.task_done()            if __name__ == '__main__':    queue = Queue()    thread_List = []    t1 = Producer(queue)    t2 = Consumer(queue)    t3 = Consumer(queue)    thread_List.append(t1)    thread_List.append(t2)    thread_List.append(t3)    for th in thread_List:        th.start()    for th in thread_List:        th.join()

import threadingimport timeclass Student(object):    def __init__(self, name, age):        self.name = name        self.age = agedef do_task1(std):    print("thread name: {}, get data {}".format(threading.current_thread().name, std.name))def do_task2(std):    print("thread name: {}, get data {}".format(threading.current_thread().name, std.name))def process_thread(name, age):    std = Student(name, age)    do_task1(std)    do_task2(std)t1 = threading.Thread(target=process_thread, args=('Alice', 21), name='Thread-A')t2 = threading.Thread(target=process_thread, args=('Bob', 22), name='Thread-B')t1.start()t2.start()t1.join()t2.join()

import threadingclass Student(object):    def __init__(self, name, age):        self.name = name        self.age = age####使用全局dict存放所有的Student对象，然后以thread自身作为key获得线程对应的Student对象global_dict = {}def process_thread(name, age):    std = Student(name, age)    global_dict[threading.currentThread()] = std # 把std放到全局变量global_dict中：    do_task1()    do_task2()def do_task1():    std = global_dict[threading.currentThread()] # 不传入std，而是根据当前线程查找：    print("thread name: {}, get data {}".format(threading.current_thread().name, std.name))def do_task2():    std = global_dict[threading.currentThread()] # 不传入std，而是根据当前线程查找：    print("thread name: {}, get data {}".format(threading.current_thread().name, std.age))t1 = threading.Thread(target=process_thread, args=('Alice', 21), name='Thread-A')t2 = threading.Thread(target=process_thread, args=('Bob', 22), name='Thread-B')t1.start()t2.start()t1.join()t2.join()

import threadingclass Student(object):    def __init__(self, name, age):        self.name = name        self.age = agelocal_school = threading.local()def do_task1():    print("thread name: {}, get data {}".format(threading.current_thread().name, local_school.std.name))def do_task2():    print("thread name: {}, get data {}".format(threading.current_thread().name, local_school.std.age))def process_thread(name, age):    std = Student(name, age)    local_school.std = std    do_task1()    do_task2()t1 = threading.Thread(target=process_thread, args=('Alice', 21), name='Thread-A')t2 = threading.Thread(target=process_thread, args=('Bob', 22), name='Thread-B')t1.start()t2.start()t1.join()t2.join()

thread name: Thread-A, get data Alicethread name: Thread-A, get data 21thread name: Thread-B, get data Bobthread name: Thread-B, get data 22

import threadingglobal_var = '123'local_thread = threading.local()local_thread.var = 'hello' ##主线程中定义本地属性varprint("global vars: {} {}".format(local_thread, local_thread.var))def do_task1():    print(global_var)    print(local_thread)    print(local_thread.var) ##子线程访问不到def do_task2():    local_thread.name = 'Bob' ##每个子线程使用全局对象local_thread，但每个线程定义的属性local_thread.name是该线程独有的。    print(global_var)    print(local_thread)    print(local_thread.name)t1 = threading.Thread(target=do_task1)t2 = threading.Thread(target=do_task2)t1.start()t2.start()

123<_thread._local object at 0x10412d8e0>BobException in thread Thread-1:Traceback (most recent call last):AttributeError: '_thread._local' object has no attribute 'var'

local_thread全局对象对主线程和子线程都是可以使用的，但是主线程定义了local_thread的属性var仅仅属于主线程所有，子线程在尝试访问属性var时，就相当于访问自己线程内的属性x，而自己线程并没有定义，就会抛出AttributeError异常：'_thread._local' object has no attribute 'var'。

import threading, timedef data_ready():    print("data is ready at {}".format(time.ctime()))if __name__ == "__main__":    th = threading.Timer(5, data_ready)    th.start()    while threading.active_count() > 1:        print("数据还没ready：{}".format(time.ctime()))        time.sleep(1)

数据还没ready：Thu Oct 18 11:59:29 2018数据还没ready：Thu Oct 18 11:59:30 2018数据还没ready：Thu Oct 18 11:59:31 2018数据还没ready：Thu Oct 18 11:59:32 2018数据还没ready：Thu Oct 18 11:59:33 2018data is ready at Thu Oct 18 11:59:34 2018

import timeimport threadingfrom concurrent.futures import ThreadPoolExecutordef data_ready(data):    time.sleep(1)    print("{} execute {} at {}".format(threading.current_thread().getname(), data, time.ctime()))    return "value_"+str(data)if __name__ == "__main__":    data_List = [1, 2, 3, 4, 5]    max_workers = 3 ##线程个数    with ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix="demo") as executor: ##max_workers参数控制线程个数，        # thread_name_prefix参数指定线程名前缀        task_List = []        for data in data_List:            task = executor.submit(data_ready, data) ##submit每次都需要提交一个目标函数和对应的参数            task_List.append(task)            task_status = task.done()  #判断task是否结束            # print(task_tag)            retry_times = 0            while task_status and retry_times < 5:                if not task_status:                    print("执行还没在执行中，waiting.....")                    time.sleep(1)                    retry_times += 1                else:                    print("任务结束了，返回值是：{}".format(task.result()))  ##返回执行结果                    break        for task in task_List:            if task.cancel():                print("cancel成功")            else:                print("已经执行")    with ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix="map_demo") as executor_map:        executor_map.map(data_ready, data_List) ##map只需要提交一次目标函数，目标函数的参数放在一个迭代器（列表，字典）里就可以。

已经执行已经执行已经执行cancel成功cancel成功 ##线程池限定线程数为3，此时demo_0/demo_1/demo_2任务已经运行，无法取消。demo_0 execute 1 at Sun Oct 21 12:18:47 2018demo_1 execute 2 at Sun Oct 21 12:18:47 2018demo_2 execute 3 at Sun Oct 21 12:18:47 2018map_demo_0 execute 1 at Sun Oct 21 12:18:48 2018map_demo_2 execute 3 at Sun Oct 21 12:18:48 2018map_demo_1 execute 2 at Sun Oct 21 12:18:48 2018map_demo_0 execute 4 at Sun Oct 21 12:18:49 2018map_demo_2 execute 5 at Sun Oct 21 12:18:49 2018

import timeimport threadingfrom concurrent.futures import ThreadPoolExecutor, as_completeddef data_ready(data):    time.sleep(1)    print("{} execute {} at {}".format(threading.current_thread().getname(), data, time.ctime()))    return "value_"+str(data)if __name__ == "__main__":    data_List = [1, 2, 3, 4, 5]    max_workers = 3 ##线程个数    with ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix="demo") as executor: ##max_workers参数控制线程个数，        task_List = [executor.submit(data_ready, data) for data in data_List]        for task in as_completed(task_List):            print("task success: {} at {}".format(task.result(), time.ctime()))

demo_0 execute 1 at Sun Oct 21 17:03:55 2018demo_1 execute 2 at Sun Oct 21 17:03:55 2018demo_2 execute 3 at Sun Oct 21 17:03:55 2018task success: value_3 at Sun Oct 21 17:03:55 2018task success: value_2 at Sun Oct 21 17:03:55 2018task success: value_1 at Sun Oct 21 17:03:55 2018demo_1 execute 5 at Sun Oct 21 17:03:56 2018demo_2 execute 4 at Sun Oct 21 17:03:56 2018task success: value_5 at Sun Oct 21 17:03:56 2018task success: value_4 at Sun Oct 21 17:03:56 2018

import timeimport threadingfrom concurrent.futures import ThreadPoolExecutor, as_completed, waitdef data_ready(data):    time.sleep(1)    print("{} execute {} at {}".format(threading.current_thread().getname(), data, time.ctime()))    return "value_"+str(data)if __name__ == "__main__":    data_List = [1, 2, 3, 4, 5]    max_workers = 3 ##线程个数    with ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix="demo") as executor: ##max_workers参数控制线程个数，        task_List = [executor.submit(data_ready, data) for data in data_List]        wait(task_List, return_when="ALL_COMPLETED") ##所有的任务完成以后再继续往下执行        print("all task is ready")        for task in as_completed(task_List):            print("task success: {} at {}".format(task.result(), time.ctime()))

demo_1 execute 2 at Sun Oct 21 17:21:28 2018demo_0 execute 1 at Sun Oct 21 17:21:28 2018demo_2 execute 3 at Sun Oct 21 17:21:28 2018demo_0 execute 4 at Sun Oct 21 17:21:29 2018demo_1 execute 5 at Sun Oct 21 17:21:29 2018all task is readytask success: value_3 at Sun Oct 21 17:21:29 2018task success: value_1 at Sun Oct 21 17:21:29 2018task success: value_5 at Sun Oct 21 17:21:29 2018task success: value_4 at Sun Oct 21 17:21:29 2018task success: value_2 at Sun Oct 21 17:21:29 2018