4月26日 python学习总结 JoinableQueue、线程、三种锁

时间 2019-11-19

标签 4月 26日 python 学习总结 joinablequeue 线程三种栏目 Python 繁體版

原文原文链接

1、进程队列补充-建立进程队列的另外一个类JoinableQueue

JoinableQueue一样经过multiprocessing使用。python

建立队列的另一个类：dom

JoinableQueue([maxsize])：这就像是一个Queue对象，但队列容许项目的使用者通知生成者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的。ide

参数介绍：函数

maxsize是队列中容许最大项数，省略则无大小限制。

方法介绍：

JoinableQueue的实例p除了与Queue对象相同的方法以外还具备：

q.task_done()：使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。若是调用此方法的次数大于从队列中删除项目的数量，将引起ValueError异常

q.join():生产者调用此方法进行阻塞，直到队列中全部的项目均被处理。阻塞将持续到队列中的每一个项目均调用q.task_done（）方法为止

 1 import time
 2 import random
 3 from multiprocessing import Process,JoinableQueue
 4 
 5 def consumer(name,q):
 6     while True:
 7         res=q.get()
 8         if res is None:break
 9         time.sleep(random.randint(1,3))
10         print('\033[46m消费者===》%s 吃了 %s\033[0m' %(name,res))
11         q.task_done()
12 
13 def producer(name,q,food):
14     for i in range(5):
15         time.sleep(random.randint(1,2))
16         res='%s%s' %(food,i)
17         q.put(res)
18         print('\033[45m生产者者===》%s 生产了 %s\033[0m' %(name,res))
19 
20 
21 
22 if __name__ == '__main__':
23     #一、共享的盆
24     q=JoinableQueue()
25 
26 
27     #二、生产者们
28     p1=Process(target=producer,args=('egon',q,'包子'))
29     p2=Process(target=producer,args=('刘',q,'泔水'))
30     p3=Process(target=producer,args=('杨',q,'米饭'))
31 
32     #三、消费者们
33     c1=Process(target=consumer,args=('alex',q))
34     c2=Process(target=consumer,args=('梁',q))
35     c1.daemon=True
36     c2.daemon=True
37 
38     p1.start()
39     p2.start()
40     p3.start()
41     c1.start()
42     c2.start()
43 
44 
45     # 肯定生产者确确实实已经生产完毕
46     p1.join()
47     p2.join()
48     p3.join()
49     # 在生产者生产完毕后，拿到队列中元素的总个数，而后直到元素总数变为0，q.join()这一行代码才算运行完毕
50     q.join()
51     #q.join()一旦结束就意味着队列确实被取空，消费者已经确确实实把数据都取干净了
52     print('主进程结束')

生产者消费者模型例子

2、线程　　

一、什么是线程post

线程指的是一条流水线的工做过程ui

进程根本就不是一个执行单位，进程实际上是一个资源单位
一个进程内自带一个线程，线程才是执行单位spa

二、进程VS线程线程

同一进程内的线程们共享该进程内资源，不一样进程内的线程资源确定是隔离的
建立线程的开销比建立进程要小的多
一个进程的全部线程的pid都是同样的
线程之间的地位相同，没有主次之分，可是主线程又表明着主进程

　　三、建立线程的两种方式　　　　

#方式一
from threading import Thread
import time
def sayhi(name):
    print('%s say hello' %name)

if __name__ == '__main__':
    t=Thread(target=sayhi,args=('egon',))
    t.start()
    print('主线程')

#方式二
from threading import Thread
import time
class Sayhi(Thread):
    def __init__(self,name):
        super().__init__()
        self.name=name
    def run(self):
        print('%s say hello' % self.name)


if __name__ == '__main__':
    t = Sayhi('egon')
    t.start()
    print('主线程')

　　　　线程执行的结果：code

　　　　与进程不一样，是由于线程启动的代价很小，启动速度远大于进程，因此，子线程先打印，主线程才打印对象

　　四、线程的其余函数　　　　　　

from threading import Thread,current_thread,active_count,enumerate
import time,os

def task():
    print('%s is running' %current_thread().name)
    time.sleep(3)

if __name__ == '__main__':
    t1=Thread(target=task,name='第一个线程')
    t2=Thread(target=task,)
    t3=Thread(target=task,)
    t1.start()
    t2.start()
    t3.start()

    # print(t1.is_alive())  #查看线程是否存活
    print(active_count())  #查看活跃的线程数，包括主线程
    print(enumerate())     #获取当前每一个活跃的线程对象
    print('主线程',current_thread().name)   #获取当前线程名

Thread实例对象的方法
  # isAlive(): 返回线程是否活动的。
  # getName(): 返回线程名。
  # setName(): 设置线程名。

threading模块提供的一些方法：
  # threading.currentThread(): 返回当前的线程变量。
  # threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。
  # threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。

五、守护线程　　

t.daemon=true
# 设为守护线程，守护线程要在线程开始执行以前设置，不然报错

不管是进程仍是线程，都遵循：守护xxx会等待主xxx运行完毕后被销毁

须要强调的是：运行完毕并不是终止运行

1 .对主进程来讲，运行完毕指的是主进程代码运行完毕

2. 对主线程来讲，运行完毕指的是主线程所在的进程内全部非守护线程通通运行完毕，主线程才算运行完毕

　详细解释：　　　　

主进程在其代码结束后就已经算运行完毕了（守护进程在此时就被回收）,而后主进程会一直等非守护的子进程都运行完毕后回收子进程的资源(不然会产生僵尸进程)，才会结束
主线程在其余非守护线程运行完毕后才算运行完毕（守护线程在此时就被回收）。由于主线程的结束意味着进程的结束，进程总体的资源都将被回收，而进程必须保证非守护线程都运行完毕后才能结束。

from threading import Thread
import time
def sayhi(name):
    time.sleep(2)
    print('%s say hello' %name)

if __name__ == '__main__':
    t=Thread(target=sayhi,args=('egon',))
    t.setDaemon(True) #必须在t.start()以前设置
    t.start()

    print('主线程')
    print(t.is_alive())
    '''
    主线程
    True
    '''

3、死锁现象；互斥锁、递归锁、信号量

死锁现象

　　　　所谓死锁：是指两个或两个以上的进程或线程在执行过程当中，因争夺资源而形成的一种互相等待的现象，若无外力做用，它们都将没法推动下去。此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程称为死锁进程，以下就是死锁

 1 from threading import Thread,Lock,RLock
 2 import time
 3 
 4 mutexA=Lock()
 5 mutexB=Lock()
 6 
 7 class MyThread(Thread):
 8     def run(self):
 9         self.f1()
10         self.f2()
11 
12     def f1(self):
13         mutexA.acquire()
14         print('%s 拿到了A锁' %self.name)
15 
16         mutexB.acquire()
17         print('%s 拿到了B锁' %self.name)
18         mutexB.release()
19 
20         mutexA.release()
21 
22     def f2(self):
23         mutexB.acquire()
24         print('%s 拿到了B锁' %self.name)
25         time.sleep(0.1)
26 
27         mutexA.acquire()
28         print('%s 拿到了A锁' %self.name)
29         mutexA.release()
30 
31         mutexB.release()
32 
33 
34 if __name__ == '__main__':
35     for i in range(10):
36         t=MyThread()
37         t.start()
38 
39     print('主')

死锁

解决方法，递归锁，在Python中为了支持在同一线程中屡次请求同一资源，python提供了可重入锁RLock。

这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源能够被屡次require。直到一个线程全部的acquire都被release，其余的线程才能得到资源。上面的例子若是使用RLock代替Lock，则不会发生死锁：

mutexA=mutexB=threading.RLock() #一个线程拿到锁，counter加1,该线程内又碰到加锁的状况，则counter继续加1，这期间全部其余线程都只能等待，等待该线程释放全部锁，即counter递减到0为止

　　信号量

同进程的同样

Semaphore管理一个内置的计数器，
每当调用acquire()时内置计数器-1；
调用release() 时内置计数器+1；
计数器不能小于0；当计数器为0时，acquire()将阻塞线程直到其余线程调用release()。

实例：(同时只有5个线程能够得到semaphore,便可以限制最大链接数为5)：　　

# from multiprocessing import Semaphore

from threading import Thread,Semaphore,current_thread
import time,random

sm=Semaphore(5)

def go_wc():
    sm.acquire()
    print('%s 上厕所ing' %current_thread().getName())
    time.sleep(random.randint(1,3))
    sm.release()

if __name__ == '__main__':
    for i in range(23):
        t=Thread(target=go_wc)
        t.start()