Python中的进程和线程

1.多进程

1.1创建进程

Unix/Linux/Mac操作系统都可以使用

fork()

函数来创建子进程，分别在父进程和子进程内返回，例如

代码:

import os  #  导入os模块

print ('当前进程的ID是：%s' % os.getpid()) # os.getpid()返回的是进程的id不是线程

ID = os.fork()  # 创建子进程，并返回进程的id，父进程返回的是父进程的id，子进程返回的是0

if ID == 0:
print ('这是子进程，ID是：%s。。父进程ID是：%s' % (os.getpid(), os.getppid()))
else:
print ('这是父进程，ID是：%s' % os.getpid())

1
2
3
4
5
6
7
8
9
10

结果:

当前进程的ID是：1064
这是父进程，ID是：1064
这是子进程，ID是：1065。。父进程ID是：1064

1
2
3
4

在Windows中没有

fork()

调用，可以用

multiprocessing

模块的

Process

类，例如

代码:

import os

from multiprocessing import Process
import os

# 子进程要执行的代码
def run_proc(name):
print('子线程 %s ，ID是：%s' % (name, os.getpid()))

print('当前线程(父线程)的ID是： %s' % os.getpid())
p = Process(target=run_proc, args=('test',))  # 创建Process的实例，并传入子线程要执行的函数和参数
p.start()  # 子线程开始执行
p.join()  # join方法用于线程间的同步，等线程执行完毕后再往下执行
print('子线程执行完毕，回到主线程%s' % os.getpid())

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

结果:

当前线程(父线程)的ID是： 1291
子线程 test ，ID是：1292
子线程执行完毕，回到主线程1291

1
2
3
4

1.2 使用进程池Pool

如果要启动多个子进程，则可用进程池

Poll

，例如

代码:

from multiprocessing import Pool
import os
import time
import random

def child_task(name):
print('子进程 %s ID是：%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('当前进程(父进程)ID是：%s' % os.getpid())
p = Pool(4)  # 创建进程池实例，大小是4个进程
for i in range(5):  # 循环5次，每次循环都创建一个子进程，大小只有4，则第五个需要等待
p.apply_async(child_task, args=(i,))  # apply_async方法，传入子进程要执行的函数和函数参数(以元组的形式)
print('子进程循环创建完毕，正在等待子进程执行。。')
p.close()  # 关闭进程池，之后就不能添加新的进程了
p.join()  # 如果有进程池，调用join前必须调用close。（join方法，等待所有子进程执行完毕再往下执行）
print('所有进程运行完毕')

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

结果:

子进程循环创建完毕，正在等待子进程执行。。
子进程 0 ID是：1776 正在运行
子进程 1 ID是：1777 正在运行
子进程 2 ID是：1778 正在运行
子进程 3 ID是：1779 正在运行
子进程 1 运行了 0.24 秒
子进程 4 ID是：1777 正在运行
子进程 3 运行了 0.64 秒
子进程 2 运行了 0.79 秒
子进程 0 运行了 1.21 秒
子进程 4 运行了 1.56 秒
所有进程运行完毕

1
2
3
4
5
6
7
8
9
10
11
12
13

1.2 进程之间的通信

进程与进程之间通过传递对象

Queue

来通信，例如

代码:

from multiprocessing import Process, Queue
import os
import time
import random

def write(q):  # 写数据
for value in ['A', 'B', 'C']:
print('进程 %s 开始写入 %s' % (os.getpid(), value))
q.put(value)
time.sleep(random.random())  # 随机睡眠一段时间，开始写入第二个数据

def read(q):  # 读数据
while True:
value = q.get(True)
print('进程 %s 开始读出 %s' % (os.getpid(), value))

if __name__ == '__main__':
q = Queue()  # 父进程创建Queue，并传给各个子进程：
pw = Process(target=write, args=(q,))  # 传入进程要执行的函数和函数参数
pr = Process(target=read, args=(q,))

pw.start()  # 启动子进程pw，写入:
pr.start()  # 启动子进程pr，读取:(启动之后，就一直循环着尝试读取，直到被中断)
pw.join()  # 等待pw结束:
pr.terminate()  # pw进程执行结束后，就中断pr，因为pr进程里是死循环，无法等待其结束，只能强行终止:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

结果:(开启了两个进程，一个读一个写)

进程 1984 开始写入 A
进程 1985 开始读出 A
进程 1984 开始写入 B
进程 1985 开始读出 B
进程 1984 开始写入 C
进程 1985 开始读出 C

1
2
3
4
5
6
7

2.多线程

2.1 创建多线程

Python 的多线程用到的是

threading

模块，同启动进程类似，传入要执行的函数，例如.

代码:

import time, threading

def loop():  # 新线程要执行的函数
print('创建了线程 %s' % threading.current_thread().name)  # current_thread()返回当前线程的名字
for n in range(5):  # 循环五次 ， 示例代码
print('线程 %s 循环第 %s 次' % (threading.current_thread().name, n + 1))
time.sleep(1)  # 暂定1秒
print('线程 %s 结束' % threading.current_thread().name)

print('最开始线程 %s 正在执行' % threading.current_thread().name)
t = threading.Thread(target=loop, name='LoopThread')  # 传入线程要执行的函数和线程的名字，如果不指定名字，系统会有默认的线程名字
t.start()
t.join()  # 等待线程执行完毕
print('线程 %s 结束' % threading.current_thread().name)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

结果:(主线程实例的名字叫MainThread)

最开始线程 MainThread 正在执行
创建了线程 LoopThread
线程 LoopThread 循环第 1 次
线程 LoopThread 循环第 2 次
线程 LoopThread 循环第 3 次
线程 LoopThread 循环第 4 次
线程 LoopThread 循环第 5 次
线程 LoopThread 结束
线程 MainThread 结束

1
2
3
4
5
6
7
8
9
10

2.1 线程锁Lock

多进程中，各自的进程变量都是各自进程独有的，进程之间互不影响，而线程中，所有线程都可以对同一个变量进行修改，这样就容易造成数据混乱。 

代码:

import threading
import time

balance = 0  # 银行存款:

def change_it(n):
global balance  # 先存后取，结果应该为0 全局变量
balance = balance + n
balance = balance - n

def run_thread(n):
for i in range(100000):
change_it(n)

t1 = threading.Thread(target=run_thread, args=(5,))
t2 = threading.Thread(target=run_thread, args=(8,))
t1.start()
t2.start()
t1.join()
t2.join()
print(balance)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

多次运行后，结果可能就不是0了，因为两个线程交叉执行，造成了数据混乱，可用

Lock

锁来确定，某个时间只能有一个线程执行该语句，例如

代码:

lock = threading.Lock()
...

def run_thread(n): # 线程执行函数代码
for i in range(100000):
lock.acquire()  # 先获取锁
try:
change_it(n)
finally:
lock.release()  # 放在finally语句中确保一定会执行，将锁释放

1
2
3
4
5
6
7
8
9
10

线程上锁的好处就是确保某时间段内只有该线程执行，缺点就是无法并发执行线程，效率较低。另可存在多个锁，有可能造成死锁的情况

2.3 python中的多核

Python线程执行前都会有一个

GIL

锁，然后，每执行100条字节码，解释器就自动释放GIL锁，让别的线程有机会执行，因此多线程在python中，只能用一个核，若要实现多核功能，可用多进程模式

3.Python中的ThreadLocal

上面已经写过，多个线程对全局变量同时做修改时会造成数据混乱，可添加互斥所来控制同一时间只有一个线程访问全局变量，但是很多时候各个线程还有自己的私有变量，如下。

代码:

import threading

def show(num):
print ('线程 %s 的结果是: %s' % (threading.current_thread().getName(), num))

def test_add():
result = 0  # 线程私有变量
for _ in xrange(1000):  # 循环1000次 xrange返回的是生成器，range返回的是list，数据量大的时候用xrange
result += 1
show(result)  # 调用其他函数的时候，需要将私有变量传递过去

threads = []
for i in range(5):
threads.append(threading.Thread(target=test_add, name=(i + 1)))  # 创建10个线程
threads[i].start()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

结果:(各自线程有各自的私有变量)

Thread-1 的结果是: 1000
Thread-2 的结果是: 1000
Thread-3 的结果是: 1000
Thread-4 的结果是: 1000
Thread-5 的结果是: 1000

1
2
3
4
5
6

若在调用

show()

函数的时候还要传入其他私有变量，或者还有很多类似

show()

函数需要调用私有变量，这样每个函数都传递私有变量太繁琐，可在全局定义一个字典，某个线程创建的时候将线程的私有变量（key=线程实例，value=变量值）加入到字典中，这样该线程内部的其他函数调用私有变量的时候不用来回传递也可取到该值，例如

代码:

import threading

data = {} # 定义的全局字典

def show():  # 线程所调用函数不用接收参数，直接获取全局字典的值即可
thread = threading.current_thread()  # 当前线程的实例
print ('线程 %s 的结果是: %s' % (thread.getName(), data[thread]))

def test_add():
thread = threading.current_thread()  # 当前线程的实例作为字典的key值
data[thread] = 0  # 初始结果为0
for _ in xrange(1000):
data[thread] += 1
show()  # 此时再调用其他函数，可不用传递参数

threads = []
for i in range(5):
threads.append(threading.Thread(target=test_add, name=(i + 1)))  # 创建5个线程
threads[i].start()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

结果:

线程 1 的结果是: 1000
线程 2 的结果是: 1000
线程 3 的结果是: 1000
线程 4 的结果是: 1000
线程 5 的结果是: 1000

1
2
3
4
5
6

然而上边的代码并没有做到完全的线程和数据分隔开，而且，每个线程执行的时候都可以访问全局字典，即可以修改其他值，这样可能也会造成数据混乱，因此，可以使用

ThreadLocal

，其实也就是相当于全局字典，只是封装好了key值，只可以访问和修改相应的value值，例如

代码:

import threading

data = threading.local()  # 获取ThreadLocal实例

def show():
thread = threading.current_thread()  # 当前线程的实例
print ('线程 %s 的结果是: %s' % (thread.getName(), data.num))  # 获取ThreadLocal实例value值data.num(key值已经绑定)

def test_add():
data.num = 0  # 直接给THreadLocal实例添加num属性，并赋值为0,还可有其他私有变量data.a,data.b等
for _ in xrange(1000):
data.num += 1
show()

threads = []
for i in range(5):
threads.append(threading.Thread(target=test_add, name=(i + 1)))  # 创建5个线程
threads[i].start()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

结果是一样的