python自动化开发学习 进程, 线程, 协程
前言
在过去单核CPU也可以执行多任务，操作系统轮流让各个任务交替执行，任务1执行0.01秒，切换任务2，任务2执行0.01秒，在切换到任务3，这样反复执行下去，表面上看每个任务都是交替执行的，但是由于CPU速度太快，让我们觉得所有任务是在同时执行一样。真正的并行执行多任务只能在多核CPU上，但是由于任务数量远远多于CPU的核心数量，所以，操作系统也会自动把多任务轮流调度到每个核心上运行。
多任务的实现有三种方式：
- 多进程模式
- 多线程模式
- 多进程+多线程模式
要实现多任务，一般会设计成Master-Worker模式，Master负责分配任务，Worker负责执行任务，因此，多任务环境下，通常是一个Master多个Worker。如果要用多线程实现，就是主线程是Master，其他线程是Worker;如果要用多进程实现，就是主进程是Master，其他进程时Worker。
多进程的优势：稳定性高，因为一个子进程挂了，不会影响主进程和其他子进程（如果主进程挂了就都挂了，不过Master一般只负责分配任务，挂掉的概率低），著名的Apache最早就是用多进程模式。
多进程的缺点：创建进程的代价太大，在Unix/Linux系统下，用fork调用还好，在windows下开销则十分巨大，另外操作系统能同时运行的进程数有限，在内存和CPU的限制下，如果有几千个进程同时运行，操作系统连调度都会成为问题。
多线程的优势：多线程模式通常比多进程快些，但也块不到哪去，在windows下，多线程的效率比多进程高，所以微软的IIS服务器默认采用多线程模式。
多线程的缺点：致命确定是任何一个线程挂掉可能直接造成整个进程崩溃，因为多线程会共享进程的内存，由于多线程存在稳定性问题，所以IIS的稳定性不如Apache。
计算密集型 VS IO密集型
计算密集型：任务特点是要进行大量的计算，消耗CPU资源，比如计算圆周率，对视频进行高清解码等等，全部依靠CPU运算能力，这种计算密集型任务虽然也可以用多任务完成，但是任务越多，花在切花任务的时间就越多，CPU执行任务的效率就越低，所以，要高效的利用CPU，计算密集型任务同时进行的数目应当等于CPU的核心数。
IO密集型：涉及到网路，磁盘IO的任务都是IO密集型任务，这类任务的特点是CPU消耗很少，任务的大部分时间都是在等待IO操作完成（因为IO的速度远远低于CPU和内存的速度）。对于IO密集型任务，任务越多，CPU效率越高，但也有个限度。常见的大部分任务都是IO密集型任务，比如WEB应用。
最后总结：
- IO密集型任务，用多线程
- 计算密集型任务，用多进程
线程
线程是最小的执行单元，而进程由最少一个线程组成。Python的标准库提供了两个模块，_thread和threading, _thread是低级模块，threading是高级模块，内部对_thread进行了封装。
绝大多数情况下，我们只需要使用threading这个高级模块。
创建线程有两种方法（其实本质是一样的）：
- 最常用的方式
最常用简单的创建线程
#启动一个线程就是把一个函数传入并创建Thread实例，然后调用start()方法，开始执行。
import threading
def f1(arg):
print(arg)
t = threading.Thread(target=func, args=(123,)) # 创建线程。Thread是个类
t.start() # 启动线程。start()会调用了run()方法执行了f1函数。
# 输出结果
最常用简单的创建线程
加强形象版创建线程
import time, threading
def func(arg):
print("线程%s 正在运行中..." % threading.current_thread().name)
time.sleep(1)
print("%s>>>%s" % (threading.current_thread().name, arg))
time.sleep(1)
print("线程%s 关闭" % threading.current_thread().name)
print("线程%s 正在运行中..." % threading.current_thread().name)
t = threading.Thread(target=func, args=('hello world',)) # 创建一个子线程
t.start() # 启动线程
t.join() # 等待所有子线程执行完任务后，在向下运行程序
print("线程%s 关闭" % threading.current_thread().name)
# 输出结果：
线程MainThread 正在运行中...
线程Thread-1 正在运行中...
Thread-1>>>hello world
线程Thread-1 关闭
线程MainThread 关闭
"""
由于任何进程默认就会启动一个线程，我们称之为主线程，主线程又可以启动新的线程，Python的threading 模块
有个 current_thread()函数，它永远返回当前线程的实例。主线程实例的名字叫MainThread，子线程的名字可以在
创建时指定，即 t = threading.Thread(target=func, args=('hello world',), name='childthread')中的name字段，
名字仅仅是在打印时用来显示的，没有任何意义，如果不指定名字，python就会自动给线程命名为Thread-1,Thread-2...
"""
加强形象版创建线程
2. 自定义一个类，继承threading.Thread,并且自己重写一个run方法
自定义类创建线程
class MyThread(threading.Thread): # 自定义一个MyThread类，继承了父类threading.Thread
def __init__(self, func, args):
self.func = func # 函数
self.args = args # 函数参数
super(MyThread, self).__init__() # 执行了父类的构造方法
def run(self): # 自定义一个run方法
self.func(self.args)
def f2(arg):
print(arg)
obj = MyThread(f2, 123) # 创建一个对象，传入f2函数，和参数
obj.start() # start会自动调用MyThread类中的run方法
# 输出结果
自定义类创建线程
锁
多线程和多进程最大的不同在于，多进程中，同一个变量各自有一份拷贝存放在每个进程中.互相不影响，而多线程中，所有变量都由所有线程共享，所以任何一个变量都可以被任何一个线程修改，从而造成了线程之间共享数据最大的危险在于多线程同时修改一个变量时，出现错误。故引入了线程锁的概念，当多个线程同时执行acquire()时，只有一个线程能成功的获取锁，然后继续执行代码，其他线程只能等待直到获得锁为止。获得锁的线程用完后一定要释放锁，否则正在等待锁的进程将永远 等待下去，进而成为了死线程。所以建议使用try...finally语句来确保锁一定会被释放。锁的好处是确保了数据的准确性，坏处在于阻止了多线程并发执行，包含锁的代码只能以单线程模式运行，效率就大大降低了。其次，由于可以存在多个锁，不同的线程持有不同的锁，并试图获取对方持有的锁，可能会造成死锁，导致多个线程全部挂起，既不执行，也无法计数，只能靠操作系统强制终止。
GIL 全局锁
Python的线程虽然是真正的线程，但是解释器在执行代码时，有一个GIL锁（Global Interpreter Lock）,任何Python线程执行前，必须先获得GIL锁，然后每执行100条字节码，解释器就会自动释放GIL锁，让别的线程有机会执行。这个GIL锁实际上把所有线程的执行代码都给上了锁，所以多线程在Python中只能交替执行，即使100个线程跑在100核CPU上，也只能利用到1核。GIL是历史遗留为，在Python中我们可以使用多线程，但是不要指望有效利用多核，如果一定要通过多线程利用多核，那只能通过C扩展来实现，不过这样就失去了Python简单易用的特点，虽然不能利用多线程实现多核任务，但是可以通过多进程实现多核任务，多个Python进程各自有互相独立的GIL锁，互不影响。
(1)线程锁 Lock RLock
由于线程之间是进行随机调度，并且每个线程可能只执行n条执行之后，当多个线程同时修改同一个条数据时可能会出现脏数据，所以，出现了线程锁：同一时刻允许一个线程执行操作。
threading.Lock() # 单锁
threading.RLock() # 多个锁，支持一个锁嵌套一个锁，一般较多使用，支持Lock
acquire() # 加锁
release() # 解锁
线程锁示例
import threading
import time
NUM = 10
def func(arg):
global NUM
arg.acquire() # 上锁
NUM -= 1
time.sleep(1)
print('当前线程为：%s, NUM值为：%s' % (threading.current_thread().name, NUM))
arg.release() # 解锁
lock = threading.RLock() # 定义锁
for i in range(10):
t = threading.Thread(target=func, args=(lock,))
t.start()
# 输出结果
当前线程为：Thread-1, NUM值为：9
当前线程为：Thread-2, NUM值为：8
当前线程为：Thread-3, NUM值为：7
当前线程为：Thread-4, NUM值为：6
当前线程为：Thread-5, NUM值为：5
当前线程为：Thread-6, NUM值为：4
当前线程为：Thread-7, NUM值为：3
当前线程为：Thread-8, NUM值为：2
当前线程为：Thread-9, NUM值为：1
当前线程为：Thread-10, NUM值为：0
线程锁示例
(2)信号量 Semaphore
互斥锁（线程锁）同时只允许一个线程更改数据，而Semaphore是同时允许一定数量的线程更改数据，比如餐厅有3个座位，那最多允许3个人同时用餐，后面的人只能等待空位才可以用餐。
信号量示例
import threading, time
NUM = 10
def func(l):
global NUM
l.acquire() # 上锁
NUM -= 1
time.sleep(2)
print("当前线程为：%s,NUM值为:%s" % (threading.current_thread().name, NUM))
l.release() # 解锁
lock = threading.BoundedSemaphore(5) # 允许5个线程同时操作
for i in range(10):
t = threading.Thread(target=func, args=(lock,))
t.start()
# 输出结果
当前线程为：Thread-3,NUM值为:5
当前线程为：Thread-1,NUM值为:5
当前线程为：Thread-2,NUM值为:5
当前线程为：Thread-5,NUM值为:2
当前线程为：Thread-4,NUM值为:2
当前线程为：Thread-8,NUM值为:0
当前线程为：Thread-6,NUM值为:0
当前线程为：Thread-7,NUM值为:0
当前线程为：Thread-10,NUM值为:0
当前线程为：Thread-9,NUM值为:0
信号量示例
(3)事件 event
Python线程的时间用于主线程控制其他线程的执行，事件主要提供了三个方法: set, wait, clear.
事件处理的机制：
- wait : 全局定义了一个“Flag”,如果"Flag"为False, 那么当程序执行event.wait 方法时就会阻塞，如果Flag为True,那么event.wait就不在阻塞。
- clear : 将Flag 设置为False
- set : 将Flag 设置为True
事件示例
import threading
def func(i,e):
print(i)
e.wait() # 表示等待检测，如果是红灯则停，绿灯则继续运行
print(i+100)
event = threading.Event()
for i in range(10):
t = threading.Thread(target=func, args=(i, event,))
t.start()
event.clear() # 设置为红灯
inp = input('>>>')
if inp == "1": # 用户输入1
event.set() # 设置为绿灯同行
# 输出结果
1
3
5
7
9
>>>1
102
103
104
108
107
# 无序是因为线程的调度时间不一致
事件示例
(4)条件 Condition
使线程等待，只有满足某条件时，才释放N个线程。
条件示例
import threading
def func(i, con):
print(threading.current_thread().name) # 输出线程名字
con.acquire() # 上锁
con.wait() # 阻塞等待
print(i+100)
con.release() # 解锁 if __name__ == '__main__':
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()
c.notify(int(inp)) # 当满足条件时，释放N个线程
c.release()
# 输出结果
Thread-1
Thread-2
Thread-3
Thread-4
Thread-5
Thread-6
Thread-7
Thread-8
Thread-9
Thread-10
>>>1 # 释放一个线程
>>>2 # 释放两个线程
102
>>>6 # 释放六个线程
105
103
108
>>>4 # 释放四个线程，然而一共只有10个，最后只剩下1个在阻塞了
>>>q # 结束循环
Process finished with exit code 0
条件示例
示例二
import threading
def condition_func():
ret = False
inp = input('>>>')
if inp == '1':
ret = True
return ret def run(n):
con.acquire()
con.wait_for(condition_func)
print("run the thread: %s" % n)
con.release()
if __name__ == '__main__':
con = threading.Condition()
for i in range(10):
t = threading.Thread(target=run, args=(i,))
t.start()
# 输出结果
>>>1
run the thread: 0
>>>1
run the thread: 1
>>>1
run the thread: 2
>>>1
run the thread: 3
>>>1
run the thread: 4
>>>1
run the thread: 5
>>>1
run the thread: 6
>>>1
run the thread: 7
>>>1
run the thread: 8
>>>1
run the thread: 9
示例二
(5)定时器 Timer
定时器，指定n秒后执行某操作。
定时器示例
from threading import Timer
def hello():
print("hello world")
t = Timer(1, hello)
t.start() # 1秒后，将执行hello函数
# 输出结果
hello wolrd
定时器示例
线程池
在使用多线程处理任务时也不是线程越多越好，在切换线程的时候，需要切换上下文环境，依然会造成CPU的大量开销。为了解决这个问题，线程池应运而生。预先创建号一个较为优化的数量的线程，让过来的任务立刻能够使用，就形成了线程池。在Python中没有内置较好的线程池模块，需要自己编写或者使用第三方模块。
简单的线程池
#!/usr/bin/env python
# -*- coding:utf-8 -*-
# Author:Liu Jiang
import queue
import time
import threading
class MyThreadPool:
def __init__(self, maxsize=5):
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)
def task(i, pool):
print(i)
time.sleep(1)
pool.add_thread()
pool = MyThreadPool(5)
for i in range(100):
t = pool.get_thread()
obj = t(target=task, args=(i,pool))
obj.start()
简单的线程池
优化的线程池
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import queue
import threading
import contextlib
import time
StopEvent = object() # 创建空对象
class ThreadPool(object):
def __init__(self, max_num, max_task_num = None):
if max_task_num:
self.q = queue.Queue(max_task_num)
else:
self.q = queue.Queue()
self.max_num = max_num
self.cancel = False
self.terminal = False
self.generate_list = []
self.free_list = []
def run(self, func, args, callback=None):
"""
线程池执行一个任务
:param func: 任务函数
:param args: 任务函数所需参数
:param callback: 任务执行失败或成功后执行的回调函数，回调函数有两个参数1、任务函数执行状态；2、任务函数返回值（默认为None，即：不执行回调函数）
:return: 如果线程池已经终止，则返回True否则None
"""
if self.cancel:
return
if len(self.free_list) == 0 and len(self.generate_list) < self.max_num:
self.generate_thread()
w = (func, args, callback,)
self.q.put(w)
def generate_thread(self):
"""
创建一个线程
"""
t = threading.Thread(target=self.call)
t.start()
def call(self):
"""
循环去获取任务函数并执行任务函数
"""
current_thread = threading.currentThread
self.generate_list.append(current_thread)
event = self.q.get()
while event != StopEvent:
func, arguments, callback = event
try:
result = func(*arguments)
success = True
except Exception as e:
success = False
result = None
if callback is not None:
try:
callback(success, result)
except Exception as e:
pass
with self.worker_state(self.free_list, current_thread):
if self.terminal:
event = StopEvent
else:
event = self.q.get()
else:
self.generate_list.remove(current_thread)
def close(self):
"""
执行完所有的任务后，所有线程停止
"""
self.cancel = True
full_size = len(self.generate_list)
while full_size:
self.q.put(StopEvent)
full_size -= 1
def terminate(self):
"""
无论是否还有任务，终止线程
"""
self.terminal = True
while self.generate_list:
self.q.put(StopEvent)
self.q.empty()
@contextlib.contextmanager
def worker_state(self, state_list, worker_thread):
"""
用于记录线程中正在等待的线程数
"""
state_list.append(worker_thread)
try:
yield
finally:
state_list.remove(worker_thread)
# How to use
pool = ThreadPool(5)
def callback(status, result):
# status, execute action status
# result, execute action return value
pass
def action(i):
print(i)
for i in range(30):
ret = pool.run(action, (i,), callback)
time.sleep(5)
print(len(pool.generate_list), len(pool.free_list))
print(len(pool.generate_list), len(pool.free_list))
# pool.close()
# pool.terminate()
优化的线程池
队列
队列都是在进程的内存中存放的，如果进程结束了，队列也就消失了。
队列的分类：
- 先进先出 : queue.Queue
- 后进先出 : queue.LifoQueue
- 权重队列 : queue.PriorityQueue
- 双向队列 : queue.deque
（1）先进先出队列
故名思议，就是先放入队列的数据，会被优先取出。
|
1
2
3
4
5
6
7
8
9
10
11
|
# 队列中的主要函数说明# put(item,block=True,timeout=None) 放数据，默认阻塞，设置条件:是否阻塞，超时时间# get(block=True,timeout=None)取数据，默认阻塞，设置条件:是否阻塞，超时时间# put_nowait(item) 相当于put(item,False)# get_nowait() 相当于get(False)# qsize() 队列中的真实个数，返回队列的大小# full() 如果队列满了，返回True，反之False# maxsize 最大支持的个数,与full()的大小相对应# empty() 检查队列是否为空，空则输出True,否则为False# join() 阻塞，等到队列为空，再执行别的操作# task_down() 在完成一项任务后，向任务已完成队列发送一个信号 |
先进先出队列示例
import queue
q = queue.Queue(5)
q.put(1)
q.put(2)
print("取出数据:%s" % q.get())
q.task_done()
print("取出数据:%s" % q.get())
q.task_done()
q.join()
print("检查队列是否为空：%s" % q.empty())
print("队列中的个数:%s" % q.qsize())
print("队列中可以存放的最大个数:%s" % q.maxsize)
# 输出结果
取出数据:1
取出数据:2
检查队列是否为空：True
队列中的个数:0
队列中可以存放的最大个数:5
先进先出队列示例
（2）后进先出队列
后进先出队列示例
import queue
q = queue.LifoQueue()
q.put(1)
q.put(2)
print(q.get())
q.task_done()
print(q.get())
q.task_done()
q.join() # 如果不使用join(),则无需使用task_done()
# 输出结果
1
后进先出队列示例
（3）权重队列
每个放入队列的内容都有一个优先级，按照优先级提取，权重值越小优先级越高，权重可以为负数。
权重队列示例
import queue
q = queue.PriorityQueue()
q.put((1, 'test1'))
q.put((0, 'test2'))
q.put((-1, 'test3'))
q.put((6, 'test4'))
q.put((2, 'test5'))
print(q.get())
# 输出结果
（-1， ‘test3’）
# 如果有优先级相等的，则按照放入顺序取出，先放先出在相同优先级的情况下
权重队列示例
（4）双向队列
|
1
2
3
4
|
append() 是向队列最右端添加appendleft() 是向队列最左端添加pop()是从队列右向左取值popleft()是从队列左向右取值 |
双向队列示例
import queue
q = queue.deque()
q.append(1)
q.append(2)
q.appendleft(3)
q.pop() # 取出队列中的一个
q.popleft() # 从队列的左边取出一个
# 输出结果
3
# 解释:
队列内容= 3 1 2
双向队列示例
简单的生产消费者模型示例
# 举例包子铺模式
import queue
import threading
import time
q = queue.Queue() def productor(arg):
"""
厨师制作的包子
"""
while True:
q.put(str(arg) + '厨师做的包子') def consumer(arg):
"""
消费者消耗包子
"""
while True:
print(arg, q.get())
time.sleep(2)
# 300用户循环消费包子
for i in range(300):
t = threading.Thread(target=consumer, args=(i,))
t.start()
# 3个出师循环制造包子
for j in range(3):
t = threading.Thread(target=productor, args=(j,))
t.start()
# 输出结果
0厨师做的包子
2厨师做的包子
1厨师做的包子
0厨师做的包子
2厨师做的包子
1厨师做的包子
0厨师做的包子
2厨师做的包子
2厨师做的包子
0厨师做的包子
1厨师做的包子
简单的生产消费者模型示例
进程
Unix/Liunx系统提供了一个fork()系统调用，它非常的特殊，普通的函数调用，调用一次，返回一次，但是fork()调用一次返回两次，因为操作系统自动把当前进程（称为父进程）复制了一份（称为子进程），
然后分别在父进程和子进程内返回。子进程永远返回0，而父进程返回子进程的ID。这样做的理由是：一个父进程可以fork出多个子进程，所以，父进程要记录下每个子进程的ID，而子进程只需要调用getppid()
就可以拿到父进程的ID。Python的os模块封装了常见的系统调用，其中包括fork，可以在Python程序中轻松创建子进程。
每个进程至少有一个线程，线程是最小的执行单元。
简单的进程示例
import os
print('Process (%s) start...' % os.getpid())
# Only works on Unix/Linux/Mac:
pid = os.fork()
if pid == 0:
print('I am child process (%s) and my parent is %s.' % (os.getpid(), os.getppid()))
else:
print('I (%s) just created a child process (%s).' % (os.getpid(), pid))
# 输出结果
Process (876) start...
I (876) just created a child process (877).
I am child process (877) and my parent is 876.
简单的进程示例
由于windows没有fork调用，所以上面代码无法在windows上运行。有了fork调用后，一个进程在接到新任务时就可以复制出一个子进程来处理新任务，常见的Apache就是由父进程监听端口，每当有新的http请求后，
就fork出一个子进程来处理新的http请求。由于Python是跨平台的，自然也有一个跨平台的多进程支持，multiprocessing模块就应运而生了。multiprocessing模块提供了一个Process类来代表一个进程对象。
示例一
from multiprocessing import Process
import os
def run_proc(name):
print("Run child process %s (%s)" % (name, os.getpid())) if __name__ == '__main__':
print('Parent process %s.' % os.getpid())
p = Process(target=run_proc, args=('test',)) # 创建子进程时，只需要出入一个执行函数和函数的参数，创建一个Process实例，启动即可。
print("Child process will start")
p.start() # 启动进程
p.join() # 等待子进程结束后在继续向下运行，通常用于进程间的同步
print("Child process end.")
# 输出结果
Parent process 1232.
Child process will start
Run child process test (7400)
Child process end..
示例一
示例二
from multiprocessing import Process
from multiprocessing import queues
import multiprocessing
def foo(i, arg):
arg.put(i)
print('say hi', i, arg.qsize()) if __name__ == '__main__':
li = queues.Queue(20, ctx=multiprocessing)
for i in range(10):
p = Process(target=foo, args=(i, li,))
p.start()
# 输出结果
say hi 0 2
say hi 1 2
say hi 2 4
say hi 3 4
say hi 4 5
say hi 5 7
say hi 6 8
say hi 7 8
say hi 8 9
say hi 9 10
示例二
注意：由于进程之间的数据需要各自持有一份，所以创建进程需要非常大的开销。
进程池：如果需要启动大量的子进程，可以通过进程池的方式批量创建子进程。
进程池示例
from multiprocessing import Pool
import os, time, random
def long_time_task(name):
print('运行子进程 %s (%s)...' % (name, os.getpid()))
start = time.time()
time.sleep(random.random() * 3)
end = time.time()
print("子进程%s运行%0.2f 秒." % (name, (end - start))) if __name__ == '__main__':
print("父进程%s." % os.getpid())
p = Pool(4)
for i in range(5):
p.apply_async(long_time_task, args=(i,))
print("等待所有子进程结束中。。。")
p.close()
p.join() # 等待所有子进程结束后才继续向下运行
print("所有子进程已经结束")
"""
调用join()之前必须先调用close(),调用close()之后，就不能继续添加新的Process了。
请注意输出结果，子进程0,1,2,3是立刻执行的，而子进程4要等待前面某个子进程完成后才
执行，这是因为Pool的进程数设置为了4，也就是最多同时执行4个进程。由于Pool的默认大小是
CPU的核数，如果你是8核CPU，要达到上述效果，你需要至少启动9个子进程。
""" # 输出结果
父进程10740.
等待所有子进程结束中。。。
运行子进程 0 (10256)...
运行子进程 1 (1172)...
运行子进程 2 (10612)...
运行子进程 3 (10292)...
子进程1运行0.11 秒.
运行子进程 4 (1172)...
子进程2运行1.09 秒.
子进程4运行1.38 秒.
子进程0运行1.79 秒.
子进程3运行2.99 秒.
所有子进程已经结束
进程池示例
子进程：很多时候，子进程并不是自身，而是一个外部进程，我们创建了子进程之后，还需要控制子进程的输入和输出。subprocess模块可以让我们非常方便地启动一个子进程，然后控制其输入和输出。
进程间通信是通过Queue,Pipes等实现的。
进程数据共享：默认进程之间无法共享数据，我们可以通过几种特殊的方法让进程间共享数据。
（1）方法一：Array
Array示例
from multiprocessing import Process
from multiprocessing import Array
def foo(i, arg):
arg[i] = i + 100
for item in arg:
print(i,"--->", item)
print('==========')
if __name__ == '__main__':
li = Array('i', 4)
for i in range(4):
p = Process(target=foo, args=(i, li, ))
p.start()
# 输出结果
---> 100
---> 0
---> 0
---> 0
==========
---> 100
---> 0
---> 102
---> 0
==========
---> 100
---> 101
---> 102
---> 0
==========
---> 100
---> 101
---> 102
---> 103
==========
Array示例
对于Array数组来说，括号内的"i"表示它内部的元素都是int型，代表的是类型。列表内的元素可以预先指定，也可以指定列表的长度。总之，Array在实例化时必须指定数组的数据类型户数组的大小。
|
1
2
3
4
5
6
7
|
类型对应表'c': ctypes.c_char, 'u': ctypes.c_wchar,
'b': ctypes.c_byte, 'B': ctypes.c_ubyte,
'h': ctypes.c_short, 'H': ctypes.c_ushort,
'i': ctypes.c_int, 'I': ctypes.c_uint,
'l': ctypes.c_long, 'L': ctypes.c_ulong,
'f': ctypes.c_float, 'd': ctypes.c_double
|
（2）方法二: manage.dict()
manage.dict()示例
from multiprocessing import Process
from multiprocessing import Manager
def Foo(i, dic):
dic[i] = 100 + i
print(dic.values())
if __name__ == '__main__':
manage = Manager()
dic = manage.dict()
for i in range(10):
p = Process(target=Foo, args=(i,dic))
p.start()
p.join()
# 输出结果
[100]
[100, 101]
[100, 101, 102]
[100, 101, 102, 103]
[100, 101, 102, 103, 104]
[100, 101, 102, 103, 104, 105]
[100, 101, 102, 103, 104, 105, 106]
[100, 101, 102, 103, 104, 105, 106, 107]
[100, 101, 102, 103, 104, 105, 106, 107, 108]
[100, 101, 102, 103, 104, 105, 106, 107, 108, 109]
manage.dict()示例
（3）方法三：queues.Queue
queue.Queue示例
import multiprocessing
from multiprocessing import Process
from multiprocessing import queues
def foo(i,arg):
arg.put(i)
print('The Process is ', i, "and the queue's size is ", arg.qsize())
if __name__ == "__main__":
li = queues.Queue(20, ctx=multiprocessing)
for i in range(10):
p = Process(target=foo, args=(i,li,))
p.start()
# 输出结果
The Process is 0 and the queue's size is 2
The Process is 1 and the queue's size is 3
The Process is 2 and the queue's size is 3
The Process is 3 and the queue's size is 4
The Process is 4 and the queue's size is 6
The Process is 5 and the queue's size is 6
The Process is 6 and the queue's size is 8
The Process is 7 and the queue's size is 8
The Process is 8 and the queue's size is 9
The Process is 9 and the queue's size is 10
queue.Queue示例
进程锁
和多线程的线程锁类似，在multiprocessing里也有同名的锁类，Lock, RLock, Event, Condition, Semaphore, 用法也是相同的。
进程锁示例
from multiprocessing import Process
from multiprocessing import Array
from multiprocessing import RLock, Lock, Event, Condition, Semaphore
import time
def foo(i,lis,lc):
lc.acquire()
lis[0] = lis[0] - 1
time.sleep(1)
print('say hi',lis[0])
lc.release()
if __name__ == "__main__":
# li = []
li = Array('i', 1)
li[0] = 10
lock = RLock()
for i in range(10):
p = Process(target=foo,args=(i,li,lock))
p.start()
# 输出结果
say hi 9
say hi 8
say hi 7
say hi 6
say hi 5
say hi 4
say hi 3
say hi 2
say hi 1
say hi 0
进程锁示例
进程池
python内置了一个进程池，不需要像线程池那样需要自定义，只要加载即可：from multiprocessing import Pool.
进程池内部维护了一个进程序列，当使用时，则去进程池中获取一个进程，如果进程池序列中没有可供使用的进程，那么程序就会等待，直到进程池中有可用进程为止。
进程池中主要方法：
- apply 从进程池里取一个进程并执行
- apply_async apply的异步版本
- terminate 立刻关闭进程池
- join 主进程等待所有子进程执行完成，必须在close或terminate之后
- close 等待所有进程结束后，才关闭进程池
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from multiprocessing import Pool
import time
def f1(args):
time.sleep(1)
print(args)
if __name__ == '__main__':
p = Pool(5)
for i in range(30):
p.apply_async(func=f1, args=(i,))
p.close() # 等子进程执行完毕后关闭线程池
# time.sleep(2)
# p.terminate() # 立刻关闭线程池
p.join() # 进程池中进程执行完毕后在关闭，如果注释的话，那么程序直接关闭
# 输出结果
1
3
...
29
进程池示例
协程
协程又称为“微线程”，英文名“Coroutine”。线程和进程的操作时由程序触发系统接口，最后的执行者是系统，协程的操作则是程序员。
协程存在的意义：对于多线程应用，CPU通过切片的方式来切换线程间的执行，线程切换时需要耗时（保存状态，下次继续）。协程则使用一个线程，在一个线程中规定某个代码块执行顺序。
协程的使用场景：当程序中存在大量不需要CPU的操作时（IO操作），适用于协程。
协程的优点：
- 一个线程执行！ 和多线程相比最大的优势就是协程极高的执行效率。因为子程序切换不是线程切换，而是由程序自身控制，因此，没有线程切换的开销，和多线程比，线程数越多，协程的性能优势越明显。
- 无需锁机制！因为只有一个线程，也不存在同时写变量的冲突，在协程中控制共享资源不加锁，只需要判断状态就好了，所以执行效率比多线程高很多。
协程如何利用多核CPU呢？最简单的方法是多进程+协程。
协程是通过第三方模块来实现的，主要是greenlet和gevent。本质上，gevent是对greenlet的高级封装，因此多用于gevent。
greenlet示例
from greenlet import greenlet
def test1():
print(12)
gr2.switch()
print(34)
gr2.switch()
def test2():
print(56)
gr1.switch()
print(78)
gr1 = greenlet(test1)
gr2 = greenlet(test2)
gr1.switch()
# 实际上，greenlet就是通过switch方法在不同的任务间进行切换
greenlet示例
gevent示例
from gevent import monkey; monkey.patch_all()
import gevent
import requests
def f(url):
print('GET: %s' % url)
resp = requests.get(url)
data = resp.text
print('%d bytes received from %s.' % (len(data), url))
gevent.joinall([
gevent.spawn(f, 'https://www.python.org/'),
gevent.spawn(f, 'https://www.yahoo.com/'),
gevent.spawn(f, 'https://github.com/'),
])
# 通过joinall将任务f和它的参数进行统一调度，实现单线程中的协程。
gevent示例