参考的文章为： [笔记]python multiprocessing模块
进程是"懒惰"的，本身不是可执行单元，主要是线程的容器。要使进程中的代码能够运行，需要至少拥有一个能够在这个环境中运行代码的执行单元，也就是线程。线程是操作系统调度和分配处理器时间的基本单位，负责执行包括在进程地址空间中的代码。当一个进程被创建时，操作系统会自动为之创建一个进程，成为主线程。一个进程可以包含多个线程。主线程可以根据需求在动态创建其他子线程。操作系统为每一个线程保存单独的寄存器环境和单独的堆栈，但是他们共享进程的地址空间，对象句柄，代码，数据和其它资源。线程总是在某个进程的上下文中被创建。
常用的threading方法：
| 函数名 | 作用 |
|---|---|
| stack_size() | 查看当前线程栈的大小 |
| stack_size(int) | 设置栈的大小 |
| active_count() 或者activeCount() | 查看活动线程数量 |
| current_thread() 或者currentThread() | 返回当前线程对象 |
| enumerate() | 枚举所有线程 |
| Thread | 线程类，用于创建和管理线程 |
| Event | 事件类，用于线程同步 |
| Condition | 条件类，用于线程同步 |
| Lock,RLock | 锁类，用于线程同步 |
| Semaphore | 信号量类，用于线程同步 |
| Timer | 用于在制定时间后调用一个函数的情况 |
Thread类
Thread类用于创建和管理线程对象，支持使用两种方法创建线程：
- 直接使用Thread类来示例化一个线程对象
- 继承Thread类并在派生类中重写_ init _()和run()
创建了线程对象后，可以调用其start()方法来启动，该方法自动调用该类对象的run()方法，此时该线程出于alinve状态，直到线程的run()方法运行结束。
thead对象的函数
| 函数名 | 作用 |
|---|---|
| start() | 自动调用run()方法，启动线程，执行线程代码 |
| run() | 线程代码，用于实现线程的功能与业务逻辑，可以在子类中重写该方法来自定义行为 |
| _ init() _() | 构造函数 |
| name | 用来读取或者设置线程的名字 |
| ident | 线程标识，非0或者None(线程未被启动) |
| is_aline()，isAlive() | 测试线程是否处于alive状态 |
| daemon | 布尔值，表示线程是否为守护线程 |
| join(tineout=None) | 等待线程结束或者超时结束 |
join([timeout])
阻塞当前线程，等待被调用线程结束或超时后在继续执行当前线程的后续代码。参数timeout用来制定最长等待时间，单位是秒。
from threading import Thread
import time
def func1(x,y):
for i in range(x,y):
print(i,end=' ')
print()
time.sleep(10)
t1 = Thread(target=func1,args=(15, 110))
t1.start()
t1.join(2)#删除该句则直接执行下一句，如果不删除会等待2秒，没有参数会等待10秒
t2 = Thread(target=func1, args=(5, 10))
t2.start()
isAlive()
用于测试线程是否处于运行状态，如果仍在运行则返回True，否则返回False。
from threading import Thread
import time
def func1():
time.sleep(10)
t1 = Thread(target=func1)
print('t1:', t1.is_alive())#程序没有运行，false
t1.start()
print('t1:', t1.is_alive())#程序还在运行，true
t1.join(5)#join因为超时而结束
print('t1:', t1.is_alive())#程序还在运行
t1.join()#等待程序结束
print('t1:', t1.is_alive())#程序运行已经结束
daemon属性
在程序运行中有一个主线程，如果在主线程中创建了子线程，当主线程结束时会根据子线程的daemon属性进行如下选择：
- 如果某个子线程的daemon属性为False主线程在结束时会检测该子线程是否结束，如果该子线程还在运行，主线程会等待该子线程结束后再退出。
- 如果某个子线程的daemon属性为True，主线程在结束时不会带这个子线程进行检查而是直接退出。
daemon属性默认为False，如果需要修改，必须在调用start()方法启动线程之前设置。```
from threading import Thread
import time
class myThread(Thread):
def __init__(self,num,threadname):
Thread.__init__(self,name=threadname)
self.num = num
def run(self):
time.sleep(self.num)
print(self.num)
t1 = myThread(1,'t1')
t2 = myThread(5,'t2')
t2.daemon = True#不会输出5，而会直接停止程序
print(t1.daemon)
print(t2.daemon)
t1.start()
t2.start()
线程同步
Lock和RLock对象
lock就是锁，锁定后面的代码，在多处加一个锁，按照顺序执行，一旦一个锁被锁定，在之后的程序中没有进行解锁，则在其他部分这个锁锁定的后续代码无法执行。
import threading
import time
import random
class mythread1(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
def run(self):
global x #使用全局的变量
print("int 1 befor lock1")
lock1.acquire() #锁定lock1
print("int 1 befor lock2")
lock2.acquire() #锁定lock2
x += 3
print('before lock1',x)
sleepTime = random.randint(0, 3)
time.sleep(sleepTime)
print(x)
# lock2.release()
lock1.release() #lock1开锁，但是lock2没有开锁
class mythread2(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
def run(self):
global x
print("in 2 befor lock1")
lock1.acquire()
print("in 2 befor lock2")
lock2.acquire()
x += 3
print('before lock2', x)
sleepTime = random.randint(0, 3)
time.sleep(sleepTime)
print(x)
lock2.release()
lock1.release()
lock1 = threading.Lock()
lock2 = threading.Lock()
t1 = mythread1()
t2 = mythread2()
x = 1
t1.start()
t2.start()
//结果：
# int 1 befor lock1
# int 1 befor lock2
# before lock1 4
# in 2 befor lock1
# 4
# in 2 befor lock2
通过观察结果可知，没有执行的语句是print('before lock2', x)，因此程序是卡在了lock2.acquire()，因为之前的的mythread1中的lock2没有释放。而Lock与RLock的区别在于RLock（可重入锁）是一个可以被同一个线程请求多次的同步指令。RLock使用了“拥有的线程”和“递归等级”的概念，处于锁定状态时，RLock被某个线程拥有。拥有RLock的线程可以再次调用acquire()，释放锁时需要调用release()相同次数。
可以认为RLock包含一个锁定池和一个初始值为0的计数器，每次成功调用 acquire()/release()，计数器将+1/-1，为0时锁处于未锁定状态。
Condition对象
该对象可以在某些事件触发后才处理数据或者执行特定的代码，可以用于不同线程之间的通信或者通知，以及更高级的同步。Condition对象除了具有acquire()和release()方法外，还有wait(),notify()/notify_all()等方法。
- acquire([timeout])/release(): 调用关联的锁的相应方法。
- wait([timeout]): 调用这个方法将使线程进入Condition的等待池等待通知，并释放锁。使用前线程必须已获得锁定，否则将抛出异常。
- notify(): 调用这个方法将从等待池挑选一个线程并通知，收到通知的线程将自动调用acquire()尝试获得锁定（进入锁定池）；其他线程仍然在等待池中。调用这个方法不会释放锁定。使用前线程必须已获得锁定，否则将抛出异常。
- notifyAll(): 调用这个方法将通知等待池中所有的线程，这些线程都将进入锁定池尝试获得锁定。调用这个方法不会释放锁定。使用前线程必须已获得锁定，否则将抛出异常。
#coding=utf-8
__author__ = 'Bruce_Zhou'
import threading
import time
import datetime
num = 0
con = threading.Condition()
class Gov(threading.Thread):
def __init__(self):
super(Gov, self).__init__()
def run(self):
global num
con.acquire() #获取一个锁
while True:
print("开始拉升股市")
num += 1
print("拉升了" + str(num) + "个点")
time.sleep(2)
if num == 5:
print("暂时安全！")
con.notify() #通知其他线程
print('gov wait before')
con.wait() #让本线程进入等待状态
print('gov wait after')
con.release()# 释放锁
class Consumers(threading.Thread):
def __init__(self):
super(Consumers, self).__init__()
def run(self):
global num
con.acquire()
while True:
if num > 0:
print("开始打压股市")
num -= 1
print("打压了" + str(num) + "个点")
time.sleep(2)
if num == 0:
print("你妹的！天台在哪里！")
con.notify()
print('consumer wait before')
con.wait()
print('consumer wait after')
con.release()
if __name__ == '__main__':
p = Gov()
c = Consumers()
p.start()
c.start()
Event对象
Event对象是一种简单的线程通信技术，一个线程设置Event对象，另一个额线程等待Event对象。Event对象的set()方法可以设置Event对象内部的信号标志为真，clear()方卡可以清楚Event对象内部的信号标志，将其设置为假；isSet()方法用来判断其内部信号的状态；wait()方法在其内部信号状态为真时会立即执行并返回，若Event对象的内部信号标志为假，wait()方法就会一直等待至超时或者内部信号状态为真。
import threading
class mythread(threading.Thread):
def __init__(self, threadname):
threading.Thread.__init__(self, name=threadname)
def run(self):
global myevent
if myevent.isSet(): #获取内部状态
myevent.clear() #清除Event对象内部的信号符号
print(self.getName() +' before wait')
myevent.wait() #在其内部信号状态为真时立即执行并返回，若Event对象的内部标志为假，wait()方法就一直等待至超时或者内部信号状态为真
print(self.getName() + 'set')
else:
print(self.getName() + 'not set')
myevent.set() #设置Event对象内部的信号标志为真
print(self.getName() + ' after set')
myevent = threading.Event()
myevent.set()
for i in range(10):
t = mythread(str(i))
t.start()
## 输出结果:
# 0 before wait
# 1not set
# 1 after set
# 0set
# 2 before wait
# 3not set
# 4not set
# 4 after set
# 3 after set
# 2set
# 5 before wait
# 6not set
# 7not set
# 7 after set
# 6 after set
# 8 before wait
# 9not set
# 9 after set
# 5set
# 8set
# 0 before wait
# 1not set
# 1 after set
# 0set
# 2 before wait
# 3not set
# 4not set
# 4 after set
# 3 after set
# 2set
# 5 before wait
# 6not set
# 6 after set
# 7 before wait
# 5set
# 8not set
# 8 after set
# 7set
# 9 before wait
从程序中可以看出一次设置为False，一次设置为True。那次时设置为true那次设置为false并不一定，可能同时几个线程在设置为true时进入循环。将flag设置为true。因此在使用时要额外注意。
多进程编程
一个进程是一个执行中的文件使用资源的总和，包括虚拟地址空间,代码,数据,对象句柄，环境变量和执行单元等。一个应用程序同时打开并执行多次，就会创建多个进程。
python标准库multiprocessing用来实现进程的创建和管理以及进程间的同步与数据的交换。用法和threading类似，时支持并行处理的重要模块。标准库multiprocessing同时支持本地兵法与远程兵法，有效地避免全局解释锁(Global Interpreter Lock ,GIL),可以有效的利用CPU环境，尤其时和多核或者多CPU环境。
常用函数
| 函数名 | 作用 |
|---|---|
| Process(target,args) | 创建进程，target为调用的进程，args为该函数需要的参数 |
| start() | 启动进程 |
| join() | 等待进程运行结束 |
进程间数据交换
multiprocessing中有多种共享的数据类型Queue等，也可以通过共享内存实现变量传递。
Queue
import threading
import time
import queue
class Producer(threading.Thread):
def __init__(self, threadname):
threading.Thread.__init__(self,name=threadname)
def run(self):
global myqueue
myqueue.put(self.getName())
print(self.getName(), 'put ', self.getName(), 'to queue.')
class Consumer(threading.Thread):
def __init__(self, threadname):
threading.Thread.__init__(self, name=threadname)
def run(self):
global myqueue
print(self.getName(), 'get ', myqueue.get(), 'from queue')
myqueue = queue.Queue()
plist = []
clist = []
for i in range(10):
p = Producer('Producer' + str(i))
plist.append(p)
c = Consumer('Consumer' + str(i))
clist.append(c)
for p, c in zip(plist,clist):
p.start()
p.join()
c.start()
c.join()
共享内存
from multiprocessing import Process,Value,Array
def f(n,a):
n.value = 3.1415926
for i in range(len(a)):
a[i] = a[i] * a[i]
if __name__=='__main__':
num = Value('d',0.0)
arr = Array('i', range(10))
p = Process(target=f, args=(num, arr))
p.start()
p.join()
print(num.value)
print(arr[:])
Manager
Manger对象控制一个拥有list,dict,Lock,RLock,Semaphore,BoundedSemaphore,Condition,Event,Barrier,Queue,Value,Array,Namespace等对象的服务端进程，并且允许其他进程访问这些对象。
from multiprocessing import Process, Manager
import time
def f(d, l, t):
d['name'] = 'Dong Fuguo'
d['age'] = 38
d['sex'] = 'Male'
d['affiliation'] = 'SDIBT'
l.reverse()
t.value = 3
if __name__=='__main__':
with Manager() as manager:
d = manager.dict()
l = manager.list(range(10))
t = manager.Value('i', 0)
p = Process(target=f, args=(d, l, t))
p.start()
# time.sleep(0.1)
p.join()
for item in d.items():
print(item)
print(l)
print(t.value)
线程同步技术
在需要协同工作完成大型任务时，多个进程间的同步非常重要。进程同步方法与线程同步类似。
使用Event对象进行进程同步
from multiprocessing import Process, Event
def f(e, i):
if e.is_set():
e.wait()
print('hello world', i)
e.clear()
else:
e.set()
if __name__ == '__main__':
e = Event()
for num in range(10):
Process(target=f, args=(e,num)).start()
#输出：
# hello world 1
# hello world 3
# hello world 5
# hello world 7
# hello world 9
使用Lock进行进程同步
from multiprocessing import Process, Lock
def f(l, i):
l.acquire()
try:
print('hello world', i)
finally:
l.release()
if __name__ == '__main__':
lock = Lock()
for num in range(10):
Process(target=f, args=(lock, num)).start()
#输出
# hello world 1
# hello world 2
# hello world 3
# hello world 4
# hello world 5
# hello world 6
# hello world 7
# hello world 8
# hello world 9
