Python核心编程——多线程threading和队列

Posted 2020-10-27 sunqi-tony

　　线程与进程类似不过他们是在同一个进程下执行的，并共享相同的上下文。可以将他们认为是在一个主进程或“主线程”中运行的迷你进程。

　　线程包括开始、执行顺序和结束三部分。它有一个指令指针，用于记录当前运行的上下文。当其他线程运行时，它可以被抢占（中断）和临时挂起（也称为睡眠）——这种做法叫做让步（yielding）。

　　threading模块知识概要

　　

　　threading.Thread

　　Thread 是threading模块中最重要的类之一，可以使用它来创建线程。有两种方式来创建线程：一种是通过继承Thread类，重写它的run方法；另一种是创建一个threading.Thread对象，在它的初始化函数　（__init__）中将可调用对象作为参数传入。

　　

 1 import threading
 2 import time
 3 ‘‘‘
 4 没有引入类，直接创建Thread实例，调用threading.Thread()方法
 5 ‘‘‘
 6 def loop(i,sleep_time):
 7     print(‘loop:‘,i,‘start‘)
 8     time.sleep(sleep_time)
 9     print(‘loop:‘,i,‘done‘)
10 def main():
11     sleep_time = [4,2]
12     loops = range(sleep_time.__len__())
13     thread_list = []
14     for i in loops:
15         t = threading.Thread(target=loop,args=(i,sleep_time[i]))
16         t.start()
17         thread_list.append(t)
18     for i in thread_list:
19         i.join()#主线程等待所有子线程执行完毕再继续执行
20     print(‘all thread have done!‘)
21 
22 if __name__ == "__main__":
23     main()

调用方法引入线程

 1 import threading
 2 import time
 3 ‘‘‘
 4 引入类，实现了面向对象编程的思想，可扩展性强
 5 ‘‘‘
 6 class threadFunc(object):
 7     def __init__(self,func,i,sleep_time):
 8         self.func = func
 9         self.i = i
10         self.sleep_time = sleep_time
11     def __call__(self, *args, **kwargs):#回调函数执行self.func(*args)
12         self.func(self.i,self.sleep_time)
13 def loop(i,sleep_time):
14     print(‘loop:‘, i, ‘start‘)
15     time.sleep(sleep_time)
16     print(‘loop:‘, i, ‘done‘)
17 def main():
18     loops = [4,2]
19     nloop = range(loops.__len__())
20     thread_list = []
21     for i in nloop:
22         t = threading.Thread(target=threadFunc(loop,i,loops[i]))
23         t.start()
24         thread_list.append(t)
25     for i in thread_list:
26         
27         i.join()
28     print(‘all thread have done!‘)
29 
30 if __name__ == ‘__main__‘:
31     main()

调用Thread函数实现面向对象思想

 1 import threading
 2 import time
 3 
 4 class Mythread(threading.Thread):
 5     def __init__(self,func,args,name = ‘‘):
 6 
 7         ‘‘‘
 8         官方文档
 9         If a subclass overrides the constructor, it must make sure to invoke
10         the base class constructor (Thread.__init__()) before doing anything
11         else to the thread.
12         如果子类重写构造函数，则必须确保在对线程执行任何其他操作之前调用基类构造函数(Thread._init_())。
13         ‘‘‘
14         threading.Thread.__init__(self)#继承父类的__init()__方法很重要
15         self.func = func
16         self.args = args
17         self.name = name
18     def run(self):
19         self.func(*self.args)
20 def loop(i,sleep_time):
21     print(‘loop:‘, i, ‘start‘)
22     time.sleep(sleep_time)
23     print(‘loop:‘, i, ‘done‘)
24 def main():
25     loops = [4,2]
26     thread_list = []
27     nloop = range(len(loops))
28     for i in nloop:
29         t = Mythread(loop,(i,loops[i]),loop.__name__)
30         t.start()
31         thread_list.append(t)
32     for i in thread_list:
33         i.join()
34     print(‘all have done!‘)
35 
36 
37 if __name__ == ‘__main__‘:
38     main()

通过继承实现线程的创建

　　Thread.join()

　　调用Thread.join将会使主调线程堵塞，直到被调用线程运行结束或超时。参数timeout是一个数值类型，表示超时时间，如果未提供该参数，那么主调线程将一直堵塞到被调线程结束。

 1 import threading, time
 2 def doWaiting():
 3   print ‘start waiting:‘, time.strftime(‘%H:%M:%S‘)
 4   time.sleep(3)
 5   print ‘stop waiting‘, time.strftime(‘%H:%M:%S‘)
 6 thread1 = threading.Thread(target = doWaiting)
 7 thread1.start()
 8 time.sleep(1) 
 9 #确保线程thread1已经启动
10 print ‘start join‘
11 thread1.join() 
12 #将一直堵塞，直到thread1运行结束。
13 print ‘end join‘

join()

 

　　threading.Lock与RLock

　　为什么要用锁这个控制机制？每个线程互相独立，相互之间没有任何关系，但是在同一个进程中的资源，线程是共享的，如果不进行资源的合理分配，对数据造成破坏，使得线程运行的结果不可预期。这种现象称为“线程不安全”。所以不排除两个进程同时访问一个数据的情况发生，一旦发生就会造成数据修改错误。

　　比如下面这个例子：

 1 import threading
 2 import time
 3 
 4 gl_num = 0
 5 
 6 def show(arg):
 7     global gl_num
 8     time.sleep(1)
 9     gl_num +=1
10     print (gl_num)
11 
12 
13 for i in range(10):
14     t = threading.Thread(target=show, args=(i,))
15     t.start()
16 
17 print (‘main thread stop‘)
18 运行结果如下
19 
20 main thread stop
21 123
22 
23 
24 45
25 
26  6678
27 
28 
29 
30 9
31 
32 Process finished

不加Lock，数据错误

 

 1 lock = threading.Lock()
 2 def show(arg):
 3 
 4     global gl_num
 5     lock.acquire()
 6     time.sleep(1)
 7     gl_num +=1
 8     lock.release()
 9     print (gl_num)
10 
11 thread_list = []
12 for i in range(10):
13     t = threading.Thread(target=show, args=(i,))
14     t.start()
15     thread_list.append(t)
16 
17 print (‘main thread stop‘)
18 正常输出1-10

加入Lock

　　在threading模块中，定义两种类型锁：threading.Lock和threading.RLock。它们之间有一点细微的区别，通过比较下面两段代码来说明：

 1 import threading
 2 lock = threading.Lock() 
 3 #Lock对象
 4 lock.acquire()
 5 lock.acquire() 
 6 #产生了死琐。
 7 lock.release()
 8 lock.release()
 9   
10 import threading
11 rLock = threading.RLock() 
12 #RLock对象
13 rLock.acquire()
14 rLock.acquire() 
15 #在同一线程内，程序不会堵塞。
16 rLock.release()
17 rLock.release()

Lock与RLock

　　这两种锁的主要区别是：RLock允许在同一线程中被多次acquire。而Lock却不允许这种情况。注意：如果使用RLock，那么acquire和release必须成对出现，即调用了n次acquire，必须调用n次的release才能真正释放所占用的锁。

　　threading.Condition（书中没有。。。）

　　线程的挂起与阻塞：

　　挂起：是因为我们需要进行调度然后认为的暂停某个线程的执行，我们也会主动放下线程实现线程的继续运行

　　阻塞：多半是被动的，因为资源访问的竞争，或是线程冲突。

　　阻塞的线程挂起放下后依然是阻塞的。

　　可以把Condiftion理解为一把高级的锁，它提供了比Lock, RLock更高级的功能，允许我们能够控制复杂的线程同步问题。

　　threadiong.Condition在内部维护一个锁对象（默认是RLock），可以在创建Condigtion对象的时候把锁对象作为参数传入。

　　condition方法：

　　1.acquire():线程获得锁

　　2.release():释放锁

　　3.wait():线程挂起状态，会自动释放锁。

　　4.notify():通知其他阻塞的线程获得锁，notifyAll():通知所有阻塞的线程，并不会自动释放锁

　　经典案例：捉迷藏

 1 #encoding: utf-8
 2 import threading, time
 3 
 4 
 5 class Seeker(threading.Thread):
 6     def __init__(self, cond, name):
 7         threading.Thread.__init__(self)
 8         self.cond = cond
 9         self.name = name
10 
11     def run(self):
12         time.sleep(1)  # 1.确保seeker晚于hider开始执行
13         print(1)
14         self.cond.acquire()  # 4. hider的锁释放了所以这里获得了锁
15         print (‘我把眼睛蒙上了‘)
16         self.cond.notify()  # 5.蒙上眼后通知hider，hider线程此时被唤醒并试图获取锁，但是锁还在seeker身上，所以hider被阻塞，seeker继续往下
17         self.cond.wait()  # 6. seeker锁被释放并且挂起，hider就获取锁开始继续往下运行了
18         print(2)
19         print (‘我找到你了‘)
20         self.cond.notify()  # 9.找到了之后通知hider，hider意图获取锁但不行所以被阻塞，seeker往下
21         self.cond.release()  # 10.释放锁
22         print(3)
23         print (‘我赢了‘)
24 
25 
26 class Hider(threading.Thread):
27     def __init__(self, cond, name):
28         threading.Thread.__init__(self)
29         self.cond = cond
30         self.name = name
31 
32     def run(self):
33         self.cond.acquire()  # 2.hider获取锁
34         self.cond.wait()  # 3.hider被挂起然后释放锁
35         print(4)
36         print (‘我已经藏好了‘)
37         self.cond.notify()  # 7.藏好后通知seeker，seeker意图获取锁，但是锁在hider身上所以seeker被阻塞
38         self.cond.wait()  # 8.hider被挂起，释放锁，seeker获取锁，seeker继续往下运行
39         print(5)
40         self.cond.release()# 11. 在此句之前一点，seeker释放了锁（#10），hider得到锁，随即这句hider释放锁
41         print (‘被你找到了‘)
42 
43 
44 cond = threading.Condition()
45 seeker = Seeker(cond, ‘seeker‘)
46 hider = Hider(cond, ‘hider‘)
47 seeker.start()
48 hider.start()

捉迷藏

　　threading.Event(书中没有自行补充。。)

Event实现与Condition类似的功能，不过比Condition简单一点。它通过维护内部的标识符来实现线程间的同步问题。

Event.wait() : 堵塞线程，直到Event对象内部标识位被设为True或超时（如果提供了参数timeout）。

Event.clear() : 将标志位置于false状态。

Event.set() : 设置标志位为true

Event.isSet() : 判断标志位状态

用Event模拟红绿灯： 

 1 import threading
 2 import time
 3 
 4 def car(event):
 5     while True:
 6         if event.isSet():
 7             print(‘green or yellow is on ,let`s go!‘)
 8             time.sleep(2)
 9         else:
10             print(‘red in on ,we must stop!‘)
11             time.sleep(2)
12 
13 def light(event):
14     ‘‘‘
15     红绿灯方法，红灯停，黄灯绿灯行
16     :return:
17     ‘‘‘
18     while True:
19         event.clear()
20         print("the light is red")
21         time.sleep(6)
22         event.set()
23         print("the light is green")
24         time.sleep(4)
25         print("the light is yellow")
26         time.sleep(2)
27 
28 
29 
30 
31 def main():
32     event = threading.Event()
33     t1 = threading.Thread(target=light,args=(event,))
34     t2 = threading.Thread(target=car,args=(event,))
35     t1.start()
36     t2.start()
37 
38 if __name__ == ‘__main__‘:
39     main()

Event 

 1 the light is red
 2 red in on ,we must stop!
 3 red in on ,we must stop!
 4 red in on ,we must stop!
 5 the light is green
 6 green or yellow is on ,let`s go!
 7 green or yellow is on ,let`s go!
 8 the light is yellow
 9 green or yellow is on ,let`s go!
10 the light is red
11 red in on ,we must stop!
12 red in on ,we must stop!
13 red in on ,we must stop!
14 the light is green
15 green or yellow is on ,let`s go!
16 green or yellow is on ,let`s go!
17 the light is yellow
18 green or yellow is on ,let`s go!
19 the light is red
20 red in on ,we must stop!
21 red in on ,we must stop!
22 red in on ,we must stop!
23 ...

运行结果

　　threading.BoundedSemaphore信号量

　　信号量是最古老的同步原语之一。它是一个计数器，当资源消耗时递减，当资源释放时递增。你可以认为信号量代表它们的资源可用或不可用。消耗资源使计数器递减的操作习惯上称为P() （来源于荷兰单词probeer/proberen），也称为wait、try、acquire、pend或procure。相对地，当一个线程对一个资源完成操作时，该资源需要返回资源池中。这个操作一般称为 V()（来源于荷兰单词 verhogen/verhoog），也称为 signal、increment、release、post、vacate。Python 简化了所有的命名，使用和锁的函数/方法一样的名字：acquire 和 release。信号量比锁更加灵活，因为可以有多个线程，每个线程拥有有限资源的一个实例。

 1 from threading import BoundedSemaphore,Lock,Thread
 2 import time
 3 import random
 4 
 5 baozi = BoundedSemaphore(3)
 6 lock = Lock()
 7 def producer():
 8     for i in range(random.randint(4,6)):
 9         lock.acquire()
10         try:
11             baozi.release()
12         except Exception:
13             print("包子普满啦")
14         else:
15             print(‘做了一个包子‘)
16         lock.release()
17         time.sleep(random.randint(1,6))
18 def customer():
19     for i in range(random.randint(3,7)):
20         lock.acquire()
21         try:
22             baozi.acquire(blocking=False)
23         except Exception:
24             print(‘包子铺没有包子啦‘)
25         else:
26             print(‘买了一个包子‘)
27         lock.release()
28         time.sleep(random.randint(3, 6))
29 
30 def main():
31     thread_list = []
32     pro = Thread(target=producer)
33     thread_list.append(pro)
34     cus = Thread(target=customer)
35     thread_list.append(cus)
36     for i in thread_list:
37         i.start()
38     for i in thread_list:
39         i.join()
40 if __name__ == ‘__main__‘:
41     main()
42 
43 
44 
45 
46 ‘‘‘
47 a = BoundedSemaphore(0)
48 if a.acquire(blocking=False):
49     print(‘ok‘)
50 else:
51     print("empty")
52     ‘‘‘

信号量

　　以上就是threading模块我目前所能掌握，一点点写出来的部分了。　

队列Queue 

队列是线程安全的！！！

　　首先，队列有很多种，根据进出顺序来分类，可以分成

　　　　Queue.Queue(maxsize)　　FIFO（先进先出队列）

　　　　Queue.LifoQueue(maxsize)　　LIFO（先进后出队列）

　　　　Queue.PriorityQueue(maxsize)　　为优先度越低的越先出来

　　　　如果设置的maxsize小于1，则表示队列的长度无限长

FIFO是常用的队列，其一些常用的方法有：

　　　　Queue.qsize()　　返回队列大小

　　　　Queue.empty()　　判断队列是否为空

　　　　Queue.full()　　判断队列是否满了

　　　　Queue.get([block[,timeout]])　　从队列头删除并返回一个item，block默认为True，表示当队列为空却去get的时候会阻塞线程，等待直到有有item出现为止来get出这个item。如果是False的话表明当队列为空你却去get的时候，会引发异常。在block为True的情况下可以再设置timeout参数。表示当队列为空，get阻塞timeout指定的秒数之后还没有get到的话就引发Full异常。

　　　　Queue.put(...[,block[,timeout]])　　向队尾插入一个item，同样若block=True的话队列满时就阻塞等待有空位出来再put，block=False时引发异常。同get的timeout，put的timeout是在block为True的时候进行超时设置的参数。

　　　　Queue.task_done()　　从场景上来说，处理完一个get出来的item之后，调用task_done将向队列发出一个信号，表示本任务已经完成

　　　　Queue.join()　　监视所有item并阻塞主线程，直到所有item都调用了task_done之后主线程才继续向下执行。这么做的好处在于，假如一个线程开始处理最后一个任务，它从任务队列中拿走最后一个任务，此时任务队列就空了但最后那个线程还没处理完。当调用了join之后，主线程就不会因为队列空了而擅自结束，而是等待最后那个线程处理完成了。

　　队列实现生产者消费者

 1 import threading
 2 import time
 3 import queue
 4 
 5 q = queue.Queue(maxsize=10)
 6 
 7 
 8 def producer(name):  # 生产者
 9     count = 1
10     while True:
11         q.put("包子%s" % count)
12         print("生产了包子", count)
13         count += 1
14         time.sleep(0.5)
15 
16 
17 def consumer(name):  # 消费者
18     while True:
19         print("[%s]取到[%s]并且吃了它..." % (name, q.get()))
20         time.sleep(1)
21 
22 
23 p = threading.Thread(target=producer, args=("Jerry",))
24 c1 = threading.Thread(target=consumer, args=("Tom",))
25 c2 = threading.Thread(target=consumer, args=("Tony",))
26 
27 p.start()
28 c1.start()
29 c2.start()

queue_pro_cus

　　队列与condition组合实现生产者消费者

 1 import queue
 2 import threading
 3 import time
 4 from threading import Condition
 5 class cus_thread(threading.Thread):
 6 
 7     def __init__(self,queue,con):
 8         threading.Thread.__init__(self);
 9         self.queue = queue
10         self.con = con
11     def run(self):
12         time.sleep(1)
13         q = self.queue
14         while True:
15             self.con.acquire()
16             if not q.empty():
17                 task = q.get()
18                 print(‘get task‘,task)
19                 self.con.notify()
20                 self.con.release()
21                 time.sleep(1)
22             else:
23                 print("no task")
24                 self.con.wait()
25 
26 class pro_thread(threading.Thread):
27     def __init__(self,queue,con):
28         threading.Thread.__init__(self)
29         self.queue = queue
30         self.con = con
31     def run(self):
32         q = self.queue
33         while True:
34             self.con.acquire()
35             if q.empty():
36                 q.put(‘111‘)
37                 print("add a job")
38                 self.con.notify()
39                 self.con.release()
40                 time.sleep(2)
41             else:
42                 self.con.wait()
43 
44 
45 def main():
46     con = Condition()
47     thread_list = []
48     q = queue.Queue()
49     pro = pro_thread(q,con)
50     thread_list.append(pro)
51     cus = cus_thread(q,con)
52     thread_list.append(cus)
53     for i in thread_list:
54         i.start()
55 
56 if __name__ == ‘__main__‘:
57     main()

queue_condition