多任务处理
目录
1.多任务的概念
现在，多核CPU已经非常普及了，但是，即使过去的单核CPU，也可以执行多任务。由于CPU执行代码都是顺序执行的，那么，单核CPU是怎么执行多任务的呢？
答案就是操作系统轮流让各个任务交替执行，任务1执行0.01秒，切换到任务2，任务2执行0.01秒，再切换到任务3，执行0.01秒……这样反复执行下去。表面上看，每个任务都是交替执行的，但是，由于CPU的执行速度实在是太快了，我们感觉就像所有任务都在同时执行一样。
真正的并行执行多任务只能在多核CPU上实现，但是，由于任务数量远远多于CPU的核心数量，所以，操作系统也会自动把很多任务轮流调度到每个核心上执行。
注意:
- 并发：指的是任务数多与cpu核数，通过操作系统的各种任务调度算法，实现用多个任务“一起”执行（实际上总有一些任务不在执行，因为切换任务的速度相当快，看上去一起执行而已）
- 并行：指的是任务数小于等于cpu核数，即任务真的是一起执行的
2.线程的使用
python的thread模块是比较底层的模块，python的threading模块是对thread做了一些包装的，可以更加方便的被使用
threading模块的使用
(1) 单线程执行
from time import sleep, ctimedef sing():for i in range(3):print("正在唱歌...%d" % i)sleep(1)def dance():for i in range(3):print("正在跳舞...%d" % i)sleep(1)if __name__ == '__main__':print('---主线程开始---:%s' % ctime())sing()dance()print('---主线程结束---:%s' % ctime())
输出结果:
---主线程开始---:Thu Sep 1 19:59:50 2022
正在唱歌...0
正在唱歌...1
正在唱歌...2
正在跳舞...0
正在跳舞...1
正在跳舞...2
---主线程结束---:Thu Sep 1 19:59:56 2022
(2) 多线程执行
import threading
from time import sleep, ctimedef sing():for i in range(3):print("正在唱歌...%d" % i)sleep(1)def dance():for i in range(3):print("正在跳舞...%d" % i)sleep(1)if __name__ == '__main__':print('---主线程开始---:%s' % ctime())t1 = threading.Thread(target=sing)t2 = threading.Thread(target=dance)t1.start()t2.start()print('---主线程结束---:%s' % ctime())
输出结果:
---主线程开始---:Thu Sep 1 20:01:34 2022
正在唱歌...0
正在跳舞...0
---主线程结束---:Thu Sep 1 20:01:34 2022
正在唱歌...1
正在跳舞...1
正在跳舞...2
正在唱歌...2
- 可以明显看出使用了多线程并发的操作，主线程没有被阻塞, 主线程花费的时间要短很多
- 当调用start()时，才会真正的创建线程，并且开始执行
(3) 查看当前线程的执行数量
# coding=utf-8
import threading
from time import sleep, ctimedef sing():for i in range(3):print("正在唱歌...%d" % i)sleep(1)def dance():for i in range(3):print("正在跳舞...%d" % i)sleep(1)if __name__ == '__main__':print('---主线程开始---:%s' % ctime())t1 = threading.Thread(target=sing)t2 = threading.Thread(target=dance)t1.start()t2.start()while True:length = len(threading.enumerate())print('当前运行的线程数为：%d' % length)if length <= 1:breaksleep(0.5)print('---主线程结束---:%s' % ctime())
输出结果:
---主线程开始---:Thu Apr 6 15:51:51 2023
正在唱歌...0
正在跳舞...0
当前运行的线程数为：3
当前运行的线程数为：3
正在唱歌...1
正在跳舞...1
当前运行的线程数为：3
当前运行的线程数为：3
正在唱歌...2
正在跳舞...2
当前运行的线程数为：3
当前运行的线程数为：3
当前运行的线程数为：1
---主线程结束---:Thu Apr 6 15:51:54 2023线程的run方法使用
前面的使用能够看出，通过使用threading模块能完成多任务的程序开发，为了让每个线程的封装性更完美，所以使用threading模块时，往往会定义一个新的子类class，只要继承threading.Thread就可以了，然后重写run方法
import threading
import timeclass MyThread(threading.Thread):def test(self):for i in range(3):time.sleep(1)# name属性中保存的是当前线程的名字msg = "I'm " + self.name + ' @ ' + str(i) print(msg)# 重写run方法def run(self):self.test()if __name__ == '__main__':t = MyThread(name='MyThread')t.start()
输出结果:
I'm MyThread @ 0
I'm MyThread @ 1
I'm MyThread @ 2
python的threading.Thread类有一个run方法，用于定义线程的功能函数，可以在自己的线程类中覆盖该方法。创建自己的线程实例后，通过Thread类的start方法，可以启动该线程，交给python虚拟机进行调度，当该线程获得执行的机会时，就会调用run方法执行线程。
线程的执行顺
多线程程序的执行顺序是不确定的。当执行到sleep语句时，线程将被阻塞（Blocked），到sleep结束后，线程进入就绪（Runnable）状态，等待调度，而线程调度将自行选择一个线程执行。
import threading
import timeclass MyThread(threading.Thread):def test(self):for i in range(3):time.sleep(1)msg = "I'm " + self.name + ' @ ' + str(i) # name属性中保存的是当前线程的名字print(msg)def run(self):self.test()if __name__ == '__main__':for i in range(5):t = MyThread(name=f'MyThread-{i}')t.start()
输出结果:
I'm MyThread-0 @ 0
I'm MyThread-3 @ 0
I'm MyThread-4 @ 0
I'm MyThread-1 @ 0
I'm MyThread-2 @ 0
I'm MyThread-4 @ 1
I'm MyThread-1 @ 1
I'm MyThread-2 @ 1I'm MyThread-0 @ 1
I'm MyThread-3 @ 1
I'm MyThread-2 @ 2
I'm MyThread-1 @ 2
I'm MyThread-4 @ 2
I'm MyThread-3 @ 2
I'm MyThread-0 @ 2
从代码和执行结果我们可以看出，多线程程序的执行顺序是不确定的。上面的代码中只能保证每个线程都运行完整个run函数，但是线程的启动顺序是不确定的。
多线程-共享全局变量
- 在一个进程内的所有线程共享全局变量，很方便在多个线程间共享数据
- 缺点就是，线程是对全局变量随意遂改可能造成多线程之间对全局变量的混乱（即线程非安全
(1) 多线程访问同一个全局变量
import threading
import timegl_num = 100def work1():# 标识使用的是全局变量global gl_numfor i in range(3):gl_num += 1print("----in work1, gl_num is %d---" % gl_num)def work2():print("----in work2, gl_num is %d---" % gl_num)print("---线程创建之前gl_num is %d---" % gl_num)
t1 = threading.Thread(target=work1)
t1.start()# 延时一会，保证t1线程中的事情做完
time.sleep(1)t2 = threading.Thread(target=work2)
t2.start()
输出结果:
---线程创建之前gl_num is 100---
----in work1, gl_num is 103---
----in work2, gl_num is 103---
从结果上可以看出线程1改变全局变量后, 线程2访问是修改后的值
(2) 多线程访问同一个局部的可变类型参数
import threading
import timedef work1(nums):nums.append(44)print("----in work1---", nums)def work2(nums):# 延时一会，保证t1线程中的事情做完time.sleep(1)print("----in work2---", nums)nums = [11, 22, 33]
print("---线程创建之前nums---", nums)t1 = threading.Thread(target=work1, args=(nums,))
t1.start()t2 = threading.Thread(target=work2, args=(nums,))
t2.start()
输出结果:
---线程创建之前nums--- [11, 22, 33]
----in work1--- [11, 22, 33, 44]
----in work2--- [11, 22, 33, 44]
可以看到, Thread类是可以支持args来设置指定函数的参数的, 注意args接收的是一个元组类型
多线程并发的问题思考
假设两个线程t1和t2都要对全局变量g_num(默认是0)进行加1运算，t1和t2都各对g_num加10次，g_num的最终的结果应该为20。
但是由于是多线程同时操作，有可能出现下面情况：
- 在g_num=0时，t1取得g_num=0。此时系统把t1调度为”sleeping”状态，把t2转换为”running”状态，t2也获得g_num=0
- 然后t2对得到的值进行加1并赋给g_num，使得g_num=1
- 然后系统又把t2调度为”sleeping”，把t1转为”running”。线程t1又把它之前得到的0加1后赋值给g_num。
- 这样导致虽然t1和t2都对g_num加1，但结果仍然是g_num=1, 最终导致g_num的结果不是20
下面通过案例来说明这个问题, 用2个线程同时对全局变量各加100万次的操作
import threading
import timeg_num = 0def work1(num):global g_numfor i in range(num):g_num += 1print("----in work1, g_num is %d---" % g_num)def work2(num):global g_numfor i in range(num):g_num += 1print("----in work2, g_num is %d---" % g_num)print("---线程创建之前g_num is %d---" % g_num)t1 = threading.Thread(target=work1, args=(1000000,))
t1.start()t2 = threading.Thread(target=work2, args=(1000000,))
t2.start()while len(threading.enumerate()) != 1:time.sleep(1)print("2个线程对同一个全局变量操作之后的最终结果是:%s" % g_num)
输出结果:
---线程创建之前g_num is 0---
----in work1, g_num is 1393864---
----in work2, g_num is 1464059---
2个线程对同一个全局变量操作之后的最终结果是:1464059
可以看到最终结果并不是200万, 这是不符合预期的。
互斥锁
当多个线程几乎同时修改某一个共享数据的时候，需要进行同步控制
线程同步能够保证多个线程安全访问竞争资源，最简单的同步机制是引入互斥锁。
互斥锁为资源引入一个状态：锁定/非锁定
某个线程要更改共享数据时，先将其锁定，此时资源的状态为“锁定”，其他线程不能更改；直到该线程释放资源，将资源的状态变成“非锁定”，其他的线程才能再次锁定该资源。互斥锁保证了每次只有一个线程进行写入操作，从而保证了多线程情况下数据的正确性。
threading模块中定义了Lock类，可以方便的处理锁定：
# 创建锁
mutex = threading.Lock()
# 锁定
mutex.acquire()
# 释放
mutex.release()
注意：
- 如果这个锁之前是没有上锁的，那么acquire不会堵塞。
- 如果在调用acquire对这个锁上锁之前 它已经被 其他线程上了锁，那么此时acquire会阻塞，直到这个锁被解锁为止。
- 当一个线程调用锁的acquire()方法获得锁时，锁就进入“locked”状态。
- 每次只有一个线程可以获得锁。如果此时另一个线程试图获得这个锁，该线程就会变为“blocked”状态，称为“阻塞”，直到拥有锁的线程调用锁的release()方法释放锁之后，锁进入“unlocked”状态。
- 锁进入“unlocked”状态后，线程调度程序从处于同步阻塞状态的线程中选择一个来获得锁，并使得该线程进入运行（running）状态。
下面使用互斥锁来实现同样的案例, 用2个线程同时对全局变量各加100万次的操作
import threading
import time# 全局变量
g_num = 0def test1(num):global g_numfor i in range(num):mutex.acquire() # 上锁g_num += 1mutex.release() # 解锁print("---test1---g_num=%d" % g_num)def test2(num):global g_numfor i in range(num):mutex.acquire() # 上锁g_num += 1mutex.release() # 解锁print("---test2---g_num=%d" % g_num)# 创建一个互斥锁
# 默认是未上锁的状态
mutex = threading.Lock()# 创建2个线程，让他们各自对g_num加1000000次
p1 = threading.Thread(target=test1, args=(1000000,))
p1.start()
p2 = threading.Thread(target=test2, args=(1000000,))
p2.start()# 等待计算完成
while len(threading.enumerate()) != 1:time.sleep(1)
print("2个线程对同一个全局变量操作之后的最终结果是:%s" % g_num)
输出结果:
---test1---g_num=1681750
---test2---g_num=2000000
2个线程对同一个全局变量操作之后的最终结果是:2000000
可以看到最后的结果，加入互斥锁后，其结果与预期相符。
锁的好处：
- 确保了某段关键代码只能由一个线程从头到尾完整地执行
锁的坏处：
- 阻止了多线程并发执行，包含锁的某段代码实际上只能以单线程模式执行，效率就大大地下降了
- 由于可以存在多个锁，不同的线程持有不同的锁，并试图获取对方持有的锁时，可能会造成死锁
死锁
在线程间共享多个资源的时候，如果两个线程分别占有一部分资源并且同时等待对方的资源，就会造成死锁。尽管死锁很少发生，但一旦发生就会造成应用的停止响应。下面看一个死锁的例子
# coding=utf-8
import threading
import timeclass MyThread1(threading.Thread):def run(self):# 对mutexA上锁mutexA.acquire()# mutexA上锁后，延时1秒，等待另外那个线程 把mutexB上锁print(self.name + '----do1---up----')time.sleep(1)# 此时会堵塞，因为这个mutexB已经被另外的线程抢先上锁了mutexB.acquire()print(self.name + '----do1---down----')mutexB.release()# 对mutexA解锁mutexA.release()class MyThread2(threading.Thread):def run(self):# 对mutexB上锁mutexB.acquire()# mutexB上锁后，延时1秒，等待另外那个线程 把mutexA上锁print(self.name + '----do2---up----')time.sleep(1)# 此时会堵塞，因为这个mutexA已经被另外的线程抢先上锁了mutexA.acquire()print(self.name + '----do2---down----')mutexA.release()# 对mutexB解锁mutexB.release()mutexA = threading.Lock()
mutexB = threading.Lock()if __name__ == '__main__':t1 = MyThread1()t2 = MyThread2()t1.start()t2.start()
输出结果:
案例-实现2个线程收发udp消息
主线程发送消息, 子线程接收消息
import socket
import threadingdef send(udp_socket):while True:msg = input("\\n请输入要发送的内容:")dst_ip = input("\\n请输入对方ip:")dst_port = int(input("\\n请输入要发送的端口号:"))udp_socket.sendto(msg.encode("utf-8"), (dst_ip, dst_port))def recv(udp_socket):while True:recv_msg = udp_socket.recvfrom(1024)recv_ip = recv_msg[1]recv_msg = recv_msg[0].decode("utf-8")print("\\n>>>%s:%s" % (str(recv_ip), recv_msg))def main():udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)udp_socket.bind(('127.0.0.1', 8089))# 创建接收线程t = threading.Thread(target=recv, args=(udp_socket,))t.start()# 主线程发送send(udp_socket)if __name__ == "__main__":main()
3. 进程的使用
(1) 概念
一个程序运行起来后，代码+用到的资源 称之为进程，它是操作系统分配资源的基本单元。
(2) 进程的状态
工作中，任务数往往大于cpu的核数，即一定有一些任务正在执行，而另外一些任务在等待cpu进行执行，因此导致了有了不同的状态
- 就绪态：运行的条件都已经满足，正在等在cpu执行
- 执行态：cpu正在执行其功能
- 等待态：等待某些条件满足，例如一个程序sleep了，此时就处于等待态
进程的创建-multiprocessing
multiprocessing模块就是跨平台版本的多进程模块，提供了一个Process类来代表一个进程对象，这个对象可以理解为是一个独立的进程，可以执行另外的事情。
创建子进程时，只需要传入一个待执行的函数和函数的参数即可，创建一个Process实例，用start()方法启动, 用法如下:
from multiprocessing import Process
import timedef run_proc():"""子进程要执行的代码"""while True:print("----2----")time.sleep(1)if __name__ == '__main__':p = Process(target=run_proc)p.start()while True:print("----1----")time.sleep(1)
输出结果:
----1----
----2----
----1----
----2----
.....
查看进程pid
通过os.getpid()可以获取到当前进程的id
from multiprocessing import Process
import osdef run_proc():"""子进程要执行的代码"""print('子进程运行中，pid=%d...' % os.getpid())print('子进程将要结束...')if __name__ == '__main__':print('父进程pid: %d' % os.getpid())p = Process(target=run_proc)p.start()
输出结果:
父进程pid: 3155
子进程运行中，pid=3157...
子进程将要结束...
Process语法结构介绍
Process([group [, target [, name [, args [, kwargs]]]]])
- target：如果传递了函数的引用，子进程就执行这里的代码
- args：给target指定的函数传递的参数，以元组的方式传递
- kwargs：给target指定的函数传递字典数据类型
- name：给进程设定一个名字，可以不设定
- group：指定进程组，大多数情况下用不到
Process创建的实例对象的常用方法：
- start()：启动子进程实例（创建子进程）
- is_alive()：判断进程子进程是否还在活着
- join([timeout])：等待子进程执行结束，或等待多少秒
- terminate()：不管任务是否完成，立即终止子进程
- name：当前进程的别名，默认为Process-N，N为从1开始递增的整数
- pid：当前进程的pid（进程号）
给子进程指定的函数传递参数
from multiprocessing import Process
import os
from time import sleepdef run_proc(name, age, kwargs):for i in range(5):print('子进程运行中，name= %s,age=%d ,pid=%d...' % (name, age, os.getpid()))print(kwargs)sleep(0.2)if __name__ == '__main__':p = Process(target=run_proc, args=('test', 18), kwargs={"m": 20})p.start()sleep(1) # 1秒中之后，立即结束子进程p.terminate()p.join()
输出结果:
子进程运行中，name= test,age=18 ,pid=58929...
子进程运行中，name= test,age=18 ,pid=58929...
子进程运行中，name= test,age=18 ,pid=58929...
子进程运行中，name= test,age=18 ,pid=58929...
子进程运行中，name= test,age=18 ,pid=58929...
{'m': 20}Process finished with exit code 0
进程和线程的区别
- 进程是系统进行资源分配和调度的一个独立单位.
- 线程是进程的一个实体,是CPU调度和分派的基本单位,它是比进程更小的能独立运行的基本单位.线程自己基本上不拥有系统资源,只拥有一点在运行中必不可少的资源(如程序计数器,一组寄存器和栈),但是它可与同属一个进程的其他的线程共享进程所拥有的全部资源.
- 一个程序至少有一个进程,一个进程至少有一个线程.
- 进程在执行过程中拥有独立的内存单元，而多个线程共享内存，从而极大地提高了程序的运行效率
- 线程不能够独立执行，必须依存在进程中
进程间通信-Queue
Process之间有时需要通信，操作系统提供了很多机制来实现进程间的通信。
可以使用multiprocessing模块的Queue实现多进程之间的数据传递，Queue本身是一个消息列队程序，首先用一个小实例来演示一下Queue的工作原理：
#coding=utf-8
from multiprocessing import Queue
q=Queue(3) #初始化一个Queue对象，最多可接收三条put消息
q.put("消息1")
q.put("消息2")
print(q.full()) #False
q.put("消息3")
print(q.full()) #True#因为消息列队已满下面的try都会抛出异常，第一个try会等待2秒后再抛出异常，第二个Try会立刻抛出异常
try:q.put("消息4",True,2)
except:print("消息列队已满，现有消息数量:%s"%q.qsize())try:q.put_nowait("消息4")
except:print("消息列队已满，现有消息数量:%s"%q.qsize())#推荐的方式，先判断消息列队是否已满，再写入
if not q.full():q.put_nowait("消息4")#读取消息时，先判断消息列队是否为空，再读取
if not q.empty():for i in range(q.qsize()):print(q.get_nowait())
运行结果:
False
True
消息列队已满，现有消息数量:3
消息列队已满，现有消息数量:3
消息1
消息2
消息3
说明:
初始化Queue()对象时（例如：q=Queue()），若括号中没有指定最大可接收的消息数量，或数量为负值，那么就代表可接受的消息数量没有上限（直到内存的尽头）；
- Queue.qsize()：返回当前队列包含的消息数量, 注意Mac Os上没有实现,调用会崩溃；
- Queue.empty()：如果队列为空，返回True，反之False ；
- Queue.full()：如果队列满了，返回True,反之False；
- Queue.get([block[, timeout]])：获取队列中的一条消息，然后将其从列队中移除，block默认值为True；
- 如果block使用默认值，且没有设置timeout（单位秒），消息列队如果为空，此时程序将被阻塞（停在读取状态），直到从消息列队读到消息为止，如果设置了timeout，则会等待timeout秒，若还没读取到任何消息，则抛出"Queue.Empty"异常；
- 如果block值为False，消息列队如果为空，则会立刻抛出"Queue.Empty"异常；
- Queue.get_nowait()：相当Queue.get(False)；
- Queue.put(item,[block[, timeout]])：将item消息写入队列，block默认值为True；
- 如果block使用默认值，且没有设置timeout（单位秒），消息列队如果已经没有空间可写入，此时程序将被阻塞（停在写入状态），直到从消息列队腾出空间为止，如果设置了timeout，则会等待timeout秒，若还没空间，则抛出"Queue.Full"异常；
- 如果block值为False，消息列队如果没有空间可写入，则会立刻抛出"Queue.Full"异常；
- Queue.put_nowait(item)：相当Queue.put(item, False)；
下面演示如何实现2个进程通过Queue来通信,在父进程中创建两个子进程，一个往Queue里写数据，一个从Queue里读数据：
from multiprocessing import Process, Queue
import os, time, random# 写数据进程执行的代码:
def write(q):for value in ['A', 'B', 'C']:print('Put %s to queue...' % value)q.put(value)time.sleep(random.random())# 读数据进程执行的代码:
def read(q):while True:if not q.empty():value = q.get(True)print('Get %s from queue.' % value)time.sleep(random.random())else:print("读取完毕")breakif __name__ == '__main__':# 父进程创建Queue，并传给各个子进程：q = Queue()pw = Process(target=write, args=(q,))pr = Process(target=read, args=(q,))# 启动子进程pw，写入:pw.start()# 等待pw结束:pw.join()# 启动子进程pr，读取:pr.start()pr.join()print('所有数据都写入并且读完')
输出结果:
Put A to queue...
Put B to queue...
Put C to queue...
Get A from queue.
Get B from queue.
Get C from queue.
读取完毕
所有数据都写入并且读完
进程池Pool
当需要创建的子进程数量不多时，可以直接利用multiprocessing中的Process动态成生多个进程，但如果是上百甚至上千个目标，手动的去创建进程的工作量巨大，此时就可以用到multiprocessing模块提供的Pool方法。
multiprocessing.Pool常用函数解析：
- apply_async(func[, args[, kwds]]) ：使用非阻塞方式调用func（并行执行，堵塞方式必须等待上一个进程退出才能执行下一个进程），args为传递给func的参数列表，kwds为传递给func的关键字参数列表；
- close()：关闭Pool，使其不再接受新的任务；
- terminate()：不管任务是否完成，立即终止；
- join()：主进程阻塞，等待子进程的退出， 必须在close或terminate之后使用；
初始化Pool时，可以指定一个最大进程数，当有新的请求提交到Pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到指定的最大值，那么该请求就会等待，直到池中有进程结束，才会用之前的进程来执行新的任务，请看下面的实例：
from multiprocessing import Pool
import os
import random
import timedef worker(msg):t_start = time.time()print("%s开始执行,进程号为%d" % (msg, os.getpid()))time.sleep(random.random() * 2)t_stop = time.time()print(msg, "执行完毕,耗时%0.2f" % (t_stop - t_start))if __name__ == '__main__':# 定义一个进程池，最大进程数3po = Pool(3)for i in range(0, 5):# Pool().apply_async(要调用的目标,(传递给目标的参数元祖,))# 每次循环将会用空闲出来的子进程去调用目标po.apply_async(worker, (i,))print("----start----")po.close() # 关闭进程池，关闭后po不再接收新的请求po.join() # 等待po中所有子进程执行完成，必须放在close语句之后print("-----end-----")
运行结果:
----start----
0开始执行,进程号为2774
1开始执行,进程号为2772
2开始执行,进程号为2773
1 执行完毕,耗时1.16
3开始执行,进程号为2772
0 执行完毕,耗时1.27
4开始执行,进程号为2774
2 执行完毕,耗时1.50
4 执行完毕,耗时1.24
3 执行完毕,耗时1.93
-----end-----
进程池中的Queue
如果要使用Pool创建进程，就需要使用multiprocessing.Manager()中的Queue()，而不是multiprocessing.Queue()，否则会得到一条如下的错误信息：
RuntimeError: Queue objects should only be shared between processes through inheritance.
下面的实例演示了进程池中的进程如何通信：
# -*- coding:utf-8 -*-# 修改import中的Queue为Manager
from multiprocessing import Manager, Pool
import os, timedef reader(q):print("reader启动(%s),父进程为(%s)" % (os.getpid(), os.getppid()))for i in range(q.qsize()):print("reader从Queue获取到消息：%s" % q.get(True))def writer(q):print("writer启动(%s),父进程为(%s)" % (os.getpid(), os.getppid()))for i in "hello":q.put(i)if __name__ == "__main__":print("(%s) start" % os.getpid())# 使用Manager中的Queueq = Manager().Queue()po = Pool()po.apply_async(writer, (q,))time.sleep(1) # 先让上面的任务向Queue存入数据，然后再让下面的任务开始从中取数据po.apply_async(reader, (q,))po.close()po.join()print("(%s) End" % os.getpid())
输出结果:
(11937) start
writer启动(11941),父进程为(11937)
reader启动(11944),父进程为(11937)
reader从Queue获取到消息：h
reader从Queue获取到消息：e
reader从Queue获取到消息：l
reader从Queue获取到消息：l
reader从Queue获取到消息：o
(11937) End
多进程实现文件夹copy
import multiprocessing
import os.path
import random
import timedef copy_files(queue, file_name, source_dir, dst_dir):f_read = open(source_dir + "/" + file_name, "rb")f_write = open(dst_dir + "/" + file_name, "wb")while True:time.sleep(random.random())byte_read = f_read.read(1024)if byte_read:f_write.write(byte_read)else:breakf_read.close()f_write.close()# 拷贝完后,添加到queuequeue.put(file_name)def main():source_dir = input("请输入要拷贝的文件夹名字:")if not os.path.exists(source_dir):print("目标文件夹不存在")returndst_dir = source_dir + "_副本"try:os.mkdir(dst_dir)except:pass# 获取源文件的列表file_names = os.listdir(source_dir)pool = multiprocessing.Pool(3)queue = multiprocessing.Manager().Queue()for file_name in file_names:# 添加任务pool.apply_async(copy_files, args=(queue, file_name, source_dir, dst_dir))pool.close()all_file_size = len(file_names)while True:file_name = queue.get()if file_name in file_names:file_names.remove(file_name)# 显示进度copy_rate = (all_file_size - len(file_names)) * 100 / all_file_sizeprint("\\r%.2f...(%s)" % (copy_rate, file_name) + " " * 50, end="")if copy_rate >= 100:breakif __name__ == "__main__":main()
4.协程的使用
什么是协程
协程，又称微线程，纤程。英文名Coroutine。
协程是python个中另外一种实现多任务的方式，只不过比线程更小占用更小执行单元（理解为需要的资源）。 为啥说它是一个执行单元，因为它自带CPU上下文。这样只要在合适的时机， 我们可以把一个协程 切换到另一个协程。 只要这个过程中保存或恢复 CPU上下文那么程序还是可以运行的。
通俗的理解：在一个线程中的某个函数，可以在任何地方保存当前函数的一些临时变量等信息，然后切换到另外一个函数中执行，注意不是通过调用函数的方式做到的，并且切换的次数以及什么时候再切换到原来的函数都由开发者自己确定。协程是在一个线程中的,所以是并发。
协程和线程差异
在实现多任务时, 线程切换从系统层面远不止保存和恢复 CPU上下文这么简单。 操作系统为了程序运行的高效性每个线程都有自己缓存Cache等等数据，操作系统还会帮你做这些数据的恢复操作。 所以线程的切换非常耗性能。但是协程的切换只是单纯的操作CPU的上下文，所以一秒钟切换个上百万次系统都抗的住。
简单实现协程
import timedef work1():while True:print("----work1---")# 交出cpu执行权yieldtime.sleep(0.5)def work2():while True:print("----work2---")# 交出cpu执行权yieldtime.sleep(0.5)def main():w1 = work1()w2 = work2()while True:next(w1)next(w2)
输出结果：
----work1---
----work2---
----work1---
----work2---
----work1---
----work2---
----work1---
----work2---
----work1---
----work2---
----work1---
----work2---
...省略...
协程-greenlet
为了更好使用协程来完成多任务，python中的greenlet模块对其封装，从而使得切换任务变的更加简单
使用前需要先安装模块
sudo pip3 install greenlet
实例代码如下：
from greenlet import greenletimport timedef test1():while True:print("---A--")gr2.switch()time.sleep(0.5)def test2():while True:print("---B--")gr1.switch()time.sleep(0.5)gr1 = greenlet(test1)
gr2 = greenlet(test2)def main():# 切换到gr1中运行gr1.switch()if __name__ == '__main__':main()
输出结果：
---A--
---B--
---A--
---B--
---A--
---B--
---A--
---B--
...省略...
协程-gevent
greenlet已经实现了协程，但是这个还的人工切换，是不是觉得太麻烦了，不要捉急，python还有一个比greenlet更强大的并且能够自动切换任务的模块gevent
其原理是当一个greenlet遇到IO(指的是input output 输入输出，比如网络、文件操作等)操作时，比如访问网络，就自动切换到其他的greenlet，等到IO操作完成，再在适当的时候切换回来继续执行。
由于IO操作非常耗时，经常使程序处于等待状态，有了gevent为我们自动切换协程，就保证总有greenlet在运行，而不是等待IO
安装模块
pip3 install gevent
代码示例：
import geventdef f(n):for i in range(n):print(gevent.getcurrent(), i)def main():print('主线程运行')g1 = gevent.spawn(f, 5)g2 = gevent.spawn(f, 5)g3 = gevent.spawn(f, 5)g1.join()g2.join()g3.join()print('主线程结束')if __name__ == '__main__':main()
运行结果：
主线程运行
<Greenlet at 0x23a9e8e6c20: f(5)> 0
<Greenlet at 0x23a9e8e6c20: f(5)> 1
<Greenlet at 0x23a9e8e6c20: f(5)> 2
<Greenlet at 0x23a9e8e6c20: f(5)> 3
<Greenlet at 0x23a9e8e6c20: f(5)> 4
<Greenlet at 0x23a9e8e6f80: f(5)> 0
<Greenlet at 0x23a9e8e6f80: f(5)> 1
<Greenlet at 0x23a9e8e6f80: f(5)> 2
<Greenlet at 0x23a9e8e6f80: f(5)> 3
<Greenlet at 0x23a9e8e6f80: f(5)> 4
<Greenlet at 0x23a9e8e7760: f(5)> 0
<Greenlet at 0x23a9e8e7760: f(5)> 1
<Greenlet at 0x23a9e8e7760: f(5)> 2
<Greenlet at 0x23a9e8e7760: f(5)> 3
<Greenlet at 0x23a9e8e7760: f(5)> 4
主线程结束可以看到，3个greenlet是依次运行而不是交替运行
模拟gevent切换执行
import geventdef f(n):for i in range(n):print(gevent.getcurrent(), i)# 用来模拟一个耗时操作，注意不是time模块中的sleepgevent.sleep(1)def main():print('主线程运行')g1 = gevent.spawn(f, 5)g2 = gevent.spawn(f, 5)g3 = gevent.spawn(f, 5)g1.join()g2.join()g3.join()print('主线程结束')if __name__ == '__main__':main()
运行结果：
主线程运行
<Greenlet at 0x26bbb1b6c20: f(5)> 0
<Greenlet at 0x26bbb1b6f80: f(5)> 0
<Greenlet at 0x26bbb1b7760: f(5)> 0
<Greenlet at 0x26bbb1b6c20: f(5)> 1
<Greenlet at 0x26bbb1b6f80: f(5)> 1
<Greenlet at 0x26bbb1b7760: f(5)> 1
<Greenlet at 0x26bbb1b6c20: f(5)> 2
<Greenlet at 0x26bbb1b6f80: f(5)> 2
<Greenlet at 0x26bbb1b7760: f(5)> 2
<Greenlet at 0x26bbb1b6c20: f(5)> 3
<Greenlet at 0x26bbb1b6f80: f(5)> 3
<Greenlet at 0x26bbb1b7760: f(5)> 3
<Greenlet at 0x26bbb1b6c20: f(5)> 4
<Greenlet at 0x26bbb1b6f80: f(5)> 4
<Greenlet at 0x26bbb1b7760: f(5)> 4
主线程结束
同样是顺序执行的，并且没有出现交替执行，所有任务都是并发执行的，也就是第一个任务执行后，才会到第二个任务执行。如果改成线程的方式来执行的话，可能看到的现象就是第二个任务先于第一个任务执行，因为他们是并行执行的。
实现并发下载
from gevent import monkey
import gevent
import urllib.request# 有耗时操作时需要
monkey.patch_all()def my_downLoad(url):print('GET: %s' % url)resp = urllib.request.urlopen(url)data = resp.read()print('%d bytes received from %s.' % (len(data), url))gevent.joinall([gevent.spawn(my_downLoad, 'http://www.baidu.com/'),gevent.spawn(my_downLoad, 'http://www.hao123.cn/'),gevent.spawn(my_downLoad, 'http://www.360.com/'),
])
输出结果：
GET: http://www.baidu.com/
GET: http://www.hao123.cn/
GET: http://www.360.com/
364400 bytes received from http://www.baidu.com/.
130162 bytes received from http://www.360.com/.
55238 bytes received from http://www.hao123.cn/.从上能够看到是先发送的获取baidu的相关信息，然后依次是hao123、360，但是收到数据的先后顺序不一定与发送顺序相同，这也就体现出了异步，即不确定什么时候会收到数据，顺序不一定
实现多个视频下载
from gevent import monkey
import gevent
import urllib.request# 有耗时才做时需要这一句
monkey.patch_all()def my_downLoad(file_name, url):print('GET: %s' % url)resp = urllib.request.urlopen(url)data = resp.read()with open(file_name, "wb") as f:f.write(data)print('%d bytes received from %s.' % (len(data), url))gevent.joinall([gevent.spawn(my_downLoad, "1.mp4", 'http://www.sample.com/sample1.mp4'),gevent.spawn(my_downLoad, "2.mp4", 'http://www.sample.com/sample2.mp4'),
])
输出结果：
GET: http://www.sample.com/sample1.mp4
GET: http://www.sample.com/sample2.mp4
110 bytes received from http://www.sample.com/sample1.mp4.
110 bytes received from http://www.sample.com/sample2.mp4.
