本节主要内容
- 线程补充
- 进程
- 协程
一.线程补充
1.两种使用方法
这里主要涉及两种使用方法，一种为直接使用，一种为定义自己的类然后继承使用如下：
- 直接使用如下：
import threading def f1(arg): print(arg) t = threading.Thread(target=f1,args=(123,)) ＃直接使用 t.start()
- 自定义类方法：
#2.自定义方法。 class Mythread(threading.Thread): ＃继承threading.Thread类 def __init__(self,func,args): ＃自己的构造函数 self.func = func self.args = args super(Mythread,self).__init__() ＃执行父的构造函数（如不需要刻意不执行） def run(self): ＃内部会先调用这个方法。此处可以自定义 self.func(self.args) obj = Mythread(f1,123) ＃创建对象。 obj.start()
2.python自带队列应用
queue 模块在python中提供队列的应用其中队列的种类有4种如下：
- 先去先出队列 queue.Queue(5)
- 后进先出队列 queue.LifoQueue()
- 优先级队列 queue.PriorityQueue()
- 双向队列 queue.deque()
以先进先出队列为例子，简单说明参数的使用：
##先进先出。 import queue q = queue.Queue(5) #最大放5个数据 q.put(11) ＃put：直接往队列中放数据 q.put(22) print(q.full()) #full判读是否为满，不满足最大的5个值 - 所以不满,输出值为False。 print(q.empty()) #empty判断是否为空，此处不空 输出值为FALSE print(q.get()) ＃get 在队列中获取数据（按照队列的顺序） print(q.get()) print(q.empty()) #此处空 输出的值为TRUE print(q.full()) q.put(33,block=False) ＃block 是否阻塞 q.put(33,block=True,timeout=2) ＃time 等待时间。 print(q.get()) print(q.get()) #print(q.get(block=False,)) ###--- 如过没有此值则会在这里阻塞(因为已经取空了)
如上为队列使用的主要方法。其它类型队列使用相同，如下：
##后进先出队列 import queue q = queue.LifoQueue() q.put(123) q.put(456) print(q.get()) ##输出456 ##优先级队列 import queue q = queue.PriorityQueue() q.put((1,111111)) q.put((0,222222)) #前面值越小优先级越低 q.put((5,333333)) q.put((9,444444)) print(q.get()) # ##双向队列 q = queue.deque() q.append(123) q.append(456) q.appendleft(111111) print(q.pop()) print(q.popleft()) # 输出最左边的 111111
3.实例练习－利用多线程＋队列实现一个生产者，消费者模型。
＃＃＃ import queue import threading import time q = queue.Queue(20) #先创建一个队列 def producter(arg): #创建一个函数-生产者-(将任务,提交到队列中供消费者获取) """ 买票提交订单到队列中 :param arg: :return: """ while True: q.put(str(arg) + '--票') #用队列的put方法将任务放到,队列中 for i in range(300): #模拟300个客户端来提交任务(比如买票) t = threading.Thread(target=producter,args=(i,)) #每个客户端创建一个线程去提交任务 t.start() def consumer(arg): #创建一个函数 - 消费者 -(在队列中获取任务然后处理) """ 在队列中获取订单信息然后处理 :param arg: :return: """ while True: print(arg,q.get()) #不停的在队列中获取任务。 time.sleep(2) #模拟此程序的处理的时间为两秒 for j in range(3): #模拟有3个消费者 t = threading.Thread(target=consumer,args=(j,)) t.start()
4.线程中的锁机制
- 线程锁机制的出现场景：
由于线程之间是随机调度的，并且当多个线程同时修改一条数据的时候，可能会出现脏数据所以出现了线程锁，同一时刻只允许某几个线程进程操作。
下面看一个例子没有使用锁的情况下：
#无锁的情况 import time,threading NUM = 10 #设定一个全局变量 - 供多个线程修改 def func(): #设置一个函数,来修改全局变量 global NUM NUM -= 1 time.sleep(1) ＃模拟处理用时需要1秒 print(NUM) for i in range(10): ＃创建多个线程，同时修改全局变量 t = threading.Thread(target=func,) t.start() ＃＃＃＃显示如下： 0 0 0 0 0 0 0 0 0 0 Process finished with exit code 0
分析： 此程序的目的是每次让全局变量自减去1然后打印出来，很明显输出结果并不是我们想要的。原因是当一个线程减去值后还没有输出其它的线程也获得了这个值然后又在这个基础上做了操作，。。。最后当所有线程都减掉1后值为0 然后大家一起输出最后所有的输出值都是0。下面引入线程的锁机制就可以限制这个情况。
Lock、RLock － 线程锁机制
下面上一段代码：采用锁机制
- 互斥锁
#互斥锁 － 同时一个线程运行 import time,threading NUM = 10 #设定一个全局变量 - 供多个线程修改 def func(lock): #设置一个函数,来修改全局变量 # #上锁 上锁后其它线程只能等待解锁后才能继续处理 lock.acquire() global NUM NUM -= 1 time.sleep(1) # #解锁 lock.release() print(NUM) lock = threading.RLock() # # lock = threading.BoundedSemaphore(5) #信号量 一次放过5个 for i in range(10): t = threading.Thread(target=func,args=(lock,)) t.start() ＃＃＃＃显示如下 9 8 7 6 5 4 3 2 1 0 Process finished with exit code 0
如上的代码输出的结果是我们最终想要的。依次输出了数字递减后的数字。如上的锁的方式为互斥锁，同时只能允许一个线程在工作。下面看下其它的类型：
- 信号量
Semaphore是同时允许一定数量的线程更改数据 ，比如厕所有3个坑，那最多只允许5个人上厕所，后面的人只能等里面有人出来了才能再进去。
# 信号量 import time,threading NUM = 10 #设定一个全局变量 - 供多个线程修改 def func(lock): #设置一个函数,来修改全局变量 # #上锁 lock.acquire() global NUM NUM -= 1 time.sleep(1) # #解锁 lock.release() print(NUM) lock = threading.BoundedSemaphore(5) #信号量 一次放过5个 for i in range(10): t = threading.Thread(target=func,args=(lock,)) t.start() ＃＃＃如上代码一次允许5个线程同时工作，等这5个都出来后然后后面的进程才能够进入处理。
- 事件锁（event）
python中的事件主要用户主线程来控制其它子线程，事件主要提供了3个方法：
- wait() 执行时判断是否等待
- set() 将 flag 设置为true
- clear() 将 flag 设置为false
event处理机制：全局定义了一个“Flag”，如果“Flag”值为 False，那么当程序执行 event.wait 方法时就会阻塞，如果“Flag”值为True，那么event.wait 方法时便不再阻塞。
如下代码实例：
#event判断（全部放或者不放）: import threading def func(i,e): ＃创建一个函数让线程执行 print(i) e.wait() #等待 运行到此处后直接等待，然后判断flag是否为false －默认为false print(i+100) ＃等待后的执行命令 envent = threading.Event() for i in range(10): t = threading.Thread(target=func,args=(i,envent)) t.start() envent.clear() #设置成红灯 - 默认是红灯。 （clear就相当于红灯） inp = input(">>>") if inp == "1": envent.set() #设置成绿灯 如果是set（）则就执行等待处后面的语句 ＃＃＃＃＃＃显示结果如下： 0 1 2 3 4 5 6 7 8 9 ＃执行到此处时等待 >>>1 ＃如果输入为1 这 e.set() 设置为绿灯。再执行后面的内容 100 104 106 107 103 105 109 101 102 108 Process finished with exit code 0
- 条件锁（Condition）
使得线程等待，只有当满足某个条件时才放行某几个线程任务：
第一种方式：
# 条件的方式判断是否应该放行 #1.第一种方式: con.wait() import threading #倒入线程模块 def func(i,con): #创建一个函数 - 被线程执行 print(i) con.acquire() #上锁, con.wait() #使线程等待 print(i+100) #等待条件成立了后的输出 con.release() #解锁 c = threading.Condition() for i in range(10): t = threading.Thread(target=func,args=(i,c,)) t.start() while True: inp = input(">>>") if inp == "q": break c.acquire() ＃此时已经有10个线程在排队等待放行 c.notify(int(inp)) #唤醒指定数量再等待的线程,默认参数为1个 在执行前必须是已经被锁了,否则会抛出错误 c.release()
第二种方式：
wait_for()
#第二种方式: import threading def condition(): #创建一个函数用于返回值。 ret = False r = input(">>") if r == 'true': #判断是否为true ret = True else: ret = False return ret def func(i,con): print(i) con.acquire() #先上锁 con.wait_for(condition) #等待结果 如果为true则继续执行 print(i+100) con.release() c = threading.Condition() for i in range(10): t = threading.Thread(target=func,args=(i,c,)) t.start() ＃＃＃＃显示结果如下： 0 >>1 2 3 4 5 6 7 8 9 true 100 >>
以上为线程中的锁机制。
5.附加Timer
线程中的定时器，指定n秒后去执行某个操作。就如下，等待两秒后去执行制定的函数。
# ##### timer from threading import Timer def hello(): print("hello, world") t = Timer(2, hello) t.start() # after 1 seconds, "hello, world" will be printed
6.简单实现线程池。
python中没有提供直接可用的线程池，所以要在python中使用线程池 1.使用第三放模块，2.自己动手丰衣足食。 下面是一个简单的线程池实现。
#########实现简单线程池 import queue import threading import time class ThreadPool: #创建一个类 def __init__(self,maxsize=5): ''' 构造方法 :param maxsize: 队列中最多成员数量。 ''' self.maxsize = maxsize self._q = queue.Queue(maxsize) #创建一个队列 for i in range(maxsize): self._q.put(threading.Thread) #将类放到队列中。 def get_thread(self): #在队列中拿到一个类 return self._q.get() def add_thread(self): #再次把类放到类中 self._q.put(threading.Thread) pool = ThreadPool(5) #创建对象 def task(arg,p): #创建一个函数让线程来执行 print(arg) time.sleep(1) p.add_thread() #每个线程工作完后都在把一个类归还到队列中 for i in range(100): t = pool.get_thread() #每次拿一个(当队列中拿完了就等待,其它的归还后在拿然后在执行) obj = t(target=task,args=(i,pool,)) obj.start()
以上为线程部分补充。
！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！
二.进程
1.创建多进程
创建多个进程如下方式（跟创建线程的方法类似，只是库名不同，其它使用方法也是类似的）
进程与进程之间默认数据是不共享的，各自持用一份，所以创建进程的开销是很大的。
from multiprocessing import Process ＃倒入模块 import time def foo(i): print('say hi', i) for i in range(10): p = Process(target=foo, args=(i,)) ＃创建方式同线程的创建方式 p.start() ＃＃显示如下： say hi 0 say hi 1 say hi 2 say hi 3 say hi 4 say hi 5 say hi 6 say hi 7 say hi 8 say hi 9 Process finished with exit code 0
2.实现进程间数据的共享
进程之间各自持有一份数据，默认是不共享的。我门来看一下如下的例子：
from multiprocessing import Process li = [] def foo(i): li.append(i) print('say hi', li) for i in range(10): p = Process(target=foo, args=(i,)) p.start() print('ending', li) ＃＃＃＃＃ say hi [0] say hi [1] say hi [2] say hi [3] say hi [4] say hi [5] say hi [6] say hi [7] ending [] say hi [8] say hi [9] Process finished with exit code 0 ＃由上面的显示可以看出来，各自维护着各自的数据，列表并没有累加
使用如下方法可以实现进程之间的数据共享：
- 1.queues
#第一种共享方法 queues from multiprocessing import Process from multiprocessing import queues import multiprocessing def foo(i,arg): arg.put(i) #arg.append(1) #print('say hi',i,li) print('say hi',i,arg) if __name__ == "__main__": #li = [] li = queues.Queue(10,ctx=multiprocessing) for i in range(10): p = Process(target=foo,args=(i,li,)) p.start() #p.join() ＃＃＃＃＃显示如下，arg的内存地址为相同的，可见操作了相同的数据 say hi 0 <multiprocessing.queues.Queue object at 0x101377ef0> say hi 1 <multiprocessing.queues.Queue object at 0x101377ef0> say hi 2 <multiprocessing.queues.Queue object at 0x101377ef0> say hi 3 <multiprocessing.queues.Queue object at 0x101377ef0> say hi 4 <multiprocessing.queues.Queue object at 0x101377ef0> say hi 5 <multiprocessing.queues.Queue object at 0x101377ef0> say hi 6 <multiprocessing.queues.Queue object at 0x101377ef0> say hi 7 <multiprocessing.queues.Queue object at 0x101377ef0> say hi 8 <multiprocessing.queues.Queue object at 0x101377ef0> say hi 9 <multiprocessing.queues.Queue object at 0x101377ef0> Process finished with exit code 0
- 2.Array
from multiprocessing import Process from multiprocessing import Array def foo(i,arg): arg[i] = i + 100 for item in arg: print(item) print("__++++__") if __name__ == "__main__": li = Array('i',10) for i in range(10): p = Process(target=foo,args=(i,li,)) p.start()
显示如下：
100 0 0 0 0 0 0 0 0 0 __++++__ 100 101 0 0 0 0 0 0 0 0 __++++__ 100 101 102 0 0 0 0 0 0 0 __++++__ 100 101 102 103 0 0 0 0 0 0 __++++__ 100 101 102 103 104 0 0 0 0 0 __++++__ 100 101 102 103 104 105 0 0 0 0 __++++__ 100 101 102 103 104 105 106 0 0 0 __++++__ 100 101 102 103 104 105 106 107 0 0 __++++__ 100 101 102 103 104 105 106 107 108 0 __++++__ 100 101 102 103 104 105 106 107 108 109 __++++__ Process finished with exit code 0 ＃＃＃可以看到所有的进程操作的都是一个数据