8.0 Python 使用进程与线程
python 进程与线程是并发编程的两种常见方式。进程是操作系统中的一个基本概念，表示程序在操作系统中的一次执行过程，拥有独立的地址空间、资源、优先级等属性。线程是进程中的一条执行路径，可以看做是轻量级的进程，与同一个进程中的其他线程共享相同的地址空间和资源。
线程和进程都可以实现并发编程，但是它们之间有几点不同：
- 线程间共享进程的内存空间，但进程间的内存空间是相互独立的；
- 线程创建和销毁的开销较小，但是线程切换的开销较大；
- 进程间通信需要较为复杂的 IPC（Inter-Process Communication）机制，线程间通信则可以直接读写共享内存；
- 多进程可以充分利用多核 CPU 的性能，但是多线程受 GIL（Global Interpreter Lock）限制，只能利用单核 CPU 的性能。
在选择使用进程还是线程时，需要根据具体场景和需求进行权衡和选择。如果任务需要充分利用多核 CPU，且任务之间互不影响，可以选择多进程；如果任务之间需要共享资源和数据，可以选择多线程。同时，需要注意在 python 中使用多线程时，由于 GIL 的存在，可能无法实现真正的并行。
8.1 创建并使用线程
线程是操作系统调度的最小执行单元，是进程中的一部分，能够提高程序的效率。在python中，创建线程需要使用threading模块。该模块的实现方法是底层调用了C语言的原生函数来实现线程的创建和管理。在系统中，所有的线程看起来都是同时执行的，但实际上是由操作系统进行时间片轮转调度的。
使用函数创建线程: 创建线程并传递参数实现指定函数多线程并发,使用join方法,等待线程执行完毕后的返回结果.
import os,time
import threading
now = lambda:time.time()
def MyThread(x,y): # 定义每个线程要执行的函数体
time.sleep(5) # 睡眠5秒钟
print("传递的数据:%s,%s"%(x,y)) # 其中有两个参数,我们动态传入
if __name__ == "__main__":
ThreadPool = []
start = now()
for item in range(10): # 创建10个线程并发执行函数
thread = threading.Thread(target=MyThread,args=(item,item+1,)) # args =>函数的参数
thread.start() # 启动线程
ThreadPool.append(thread)
for item in ThreadPool:
item.join()
print("[+] 线程信息: {}".format(item))
stop = now()
print("[+] 线程总耗时: {} s".format(int(stop-start)))
使用类创建内部线程: 通过定义类,将线程函数与类进行结合实现一体化该方式调用方便思维明确.
import os,time
import threading
class MyThread(threading.Thread):
def __init__(self,x,y):
super(MyThread, self).__init__()
self.x = x
self.y = y
def run(self): # 用于执行相应操作(固定写法)
print("[+] 当前执行运算: {} + {}".format(self.x,self.y))
self.result = self.x + self.y
def get_result(self): # 获取计算结果
try:
return self.result
except Exception:
return None
if __name__ == "__main__":
ThreadPool = []
for item in range(1,10):
obj = MyThread(item,item+1)
obj.start()
ThreadPool.append(obj)
for item in ThreadPool:
item.join()
print("[+] 获取返回: ",item.get_result())
使用类创建外部线程: 该定义方式与上方完全不同,我们可以将执行过程定义到类的外部为单独函数,然后类内部去调用传参.
import os,time
import threading
def MyThreadPrint(x,y):
print("[+] 当前执行运算: {} + {}".format(x,y))
result = x + y
return result
class MyThread(threading.Thread):
def __init__(self,func,args=()):
super(MyThread, self).__init__()
self.func = func
self.args = args
def run(self):
self.result = self.func(*self.args)
def get_result(self):
try:
return self.result
except Exception:
return None
if __name__ == "__main__":
ThreadPool = []
for item in range(1,10):
obj = MyThread(func=MyThreadPrint,args=(item,item+1))
obj.start()
ThreadPool.append(obj)
for item in ThreadPool:
item.join()
print("[+] 获取返回: ",item.get_result())
在线程中创建子线程: 通过创建一个守护线程,并让守护线程调用子线程,从而实现线程中调用线程,线程嵌套调用.
import time
import threading
# run => 子线程 => 由主线程调用它
def run(num):
print("这是第 {} 个子线程".format(num))
time.sleep(2)
# main = > 主守护线程 => 在里面运行5个子线程
def main():
for each in range(5):
thread = threading.Thread(target=run,args=(each,))
thread.start()
print("启动子线程: {} 编号: {}".format(thread.getName(),each))
thread.join()
if __name__ == "__main__":
daemon = threading.Thread(target=main,args=())
daemon.setDaemon(True) # 设置主线程为守护线程
daemon.start() # 启动守护线程
daemon.join(timeout=10) # 设置10秒后关闭,不论子线程是否执行完毕
简单的线程互斥锁(Semaphore): 同时允许一定数量的线程更改数据,也就是限制每次允许执行的线程数.
import threading,time
semaphore = threading.BoundedSemaphore(5) #最多允许5个线程同时运行
def run(n):
semaphore.acquire() #添加信号
time.sleep(1)
print("运行这个线程中: %s"%n)
semaphore.release() #关闭信号
if __name__ == '__main__':
for i in range(20): #同时执行20个线程
t = threading.Thread(target=run, args=(i,))
t.start()
while threading.active_count() != 1: #等待所有线程执行完毕
pass
else:
print('----所有线程执行完毕了---')
import threading,time
class mythreading(threading.Thread):
def run(self):
semaphore.acquire() #获取信号量锁
print('running the thread:',self.getName())
time.sleep(2)
semaphore.release() #释放信号量锁
if __name__ == "__main__":
semaphore = threading.BoundedSemaphore(3) # 只运行3个线程同时运行
for i in range(20):
t1 = mythreading()
t1.start()
t1.join()
线程全局锁(Lock): 添加本全局锁以后,能够保证在同一时间内保证只有一个线程具有权限.
import time
import threading
num = 0 #定义全局共享变量
thread_list = [] #线程列表
lock = threading.Lock() #生成全局锁
def SumNumber():
global num #在每个线程中获取这个全局变量
time.sleep(2)
lock.acquire() #修改数据前给数据加锁
num += 1 #每次进行递增操作
lock.release() #执行完毕以后,解除锁定
for x in range(50): #指定生成线程数
thread = threading.Thread(target=SumNumber)
thread.start() #启动线程
thread_list.append(thread) #将结果列表加入到变量中
for y in thread_list: #等待执行完毕.
y.join()
print("计算结果: ",num)
线程递归锁(RLock): 递归锁和全局锁差不多,递归锁就是在大锁中还要添加个小锁,递归锁是常用的锁.
import threading
import time
num = 0 #初始化全局变量
lock = threading.RLock() #设置递归锁
def fun1():
lock.acquire() #添加递归锁
global num
num += 1
lock.release() #关闭递归锁
return num
def fun2():
lock.acquire() #添加递归锁
res = fun1()
print("计算结果: ",res)
lock.release() #关闭递归锁
if __name__ == "__main__":
for x in range(10): #生成10个线程
thread = threading.Thread(target=fun2)
thread.start()
while threading.active_count() != 1: #等待所有线程执行完成
print(threading.active_count())
else:
print("所有线程运行完成...")
print(num)
线程互斥锁量控制并发: 使用BoundedSemaphore定义默认信号10,既可以实现控制单位时间内的程序并发量.
import os,time
import threading
def MyThread(x):
lock.acquire() # 上锁
print("执行数据: {}".format(x))
lock.release() # 释放锁
time.sleep(2) # 模拟函数消耗时间
threadmax.release() # 释放信号,可用信号加1
if __name__ == "__main__":
# 此处的BoundedSemaphore就是说默认给与10个信号
threadmax = threading.BoundedSemaphore(10) # 限制线程的最大数量为10个
lock = threading.Lock() # 将锁内的代码串行化(防止print输出乱行)
ThreadPool = [] # 执行线程池
for item in range(1,100):
threadmax.acquire() # 增加信号,可用信号减1
thread = threading.Thread(target=MyThread,args=(item,))
thread.start()
ThreadPool.append(thread)
for item in ThreadPool:
item.join()
线程驱动事件(Event): 线程事件用于主线程控制其他线程的执行,事件主要提供了三个方法set、wait、clear、is_set,分别用于设置检测和清除标志.
import threading
event = threading.Event()
def func(x,event):
print("函数被执行了: %s 次.." %x)
event.wait() # 检测标志位状态,如果为True=继续执行以下代码,反之等待.
print("加载执行结果: %s" %x)
for i in range(10): # 创建10个线程
thread = threading.Thread(target=func,args=(i,event,))
thread.start()
print("当前状态: %s" %event.is_set()) # 检测当前状态,这里为False
event.clear() # 将标志位设置为False,默认为False
temp=input("输入yes解锁新姿势: ") # 输入yes手动设置为True
if temp == "yes":
event.set() # 设置成True
print("当前状态: %s" %event.is_set()) # 检测当前状态,这里为True
import threading
def show(event):
event.wait() # 阻塞线程执行程序
print("执行一次线程操作")
if __name__ == "__main__":
event_obj = threading.Event() # 创建event事件对象
for i in range(10):
t1 = threading.Thread(target=show,args=(event_obj,))
t1.start()
inside = input('>>>:')
if inside == '1':
event_obj.set() # 当用户输入1时set全局Flag为True,线程不再阻塞
event_obj.clear() # 将Flag设置为False
线程实现条件锁: 条件(Condition) 使得线程等待,只有满足某条件时,才释放N个线程.
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('running...',n)
con.release() # 释放锁
if __name__ == "__main__":
con = threading.Condition() # 建立线程条件对象
for i in range(10):
t = threading.Thread(target=run,args=(i,))
t.start()
t.join()
单线程异步并发执行: 在单线程下实现异步执行多个函数,返回耗时取决于最后一个函数的执行时间.
import time,asyncio
now = lambda :time.time()
async def GetSystemMem(sleep):
print("[+] 执行获取内存异步函数.")
await asyncio.sleep(sleep) # 设置等待时间
return 1
async def GetSystemCPU(sleep):
print("[+] 执行获取CPU异步函数.")
await asyncio.sleep(sleep) # 设置等待时间
return 1
if __name__ == "__main__":
stop = now()
mem = GetSystemMem(1)
cpu = GetSystemCPU(4)
task=[
asyncio.ensure_future(mem), # 将多个任务添加进一个列表
asyncio.ensure_future(cpu)
]
loop=asyncio.get_event_loop() # 创建一个事件循环
loop.run_until_complete(asyncio.wait(task)) # 开始并发执行
for item in task:
print("[+] 返回结果: ",item.result()) # 输出回调
print('总耗时: {}'.format(stop - now()))
8.2 创建并使用进程
进程是指正在执行的程序，创建进程需要使用multiprocessing模块，创建方法和线程相同，但由于进程之间的数据需要各自持有一份，所以创建进程需要更大的开销。进程间数据不共享，多进程可以用来处理多任务，但很消耗资源。计算密集型任务最好交给多进程来处理，I/O密集型任务最好交给多线程来处理。另外，进程的数量应该和CPU的核心数保持一致，以充分利用系统资源。
使用进程函数执行命令: 通过系统提供的进程线程函数完成对系统命令的调用与执行.
>>> import os,subprocess
>>>
>>> os.system("ping -n 1 www.baidu.com") # 在当前shell中执行命令
>>>
>>> ret = os.popen("ping -n 1 www.baidu.com") # 在子shell中执行命令
>>> ret.read()
>>>
>>> subprocess.run("ping www.baidu.com",shell=True)
>>> subprocess.call("ping www.baidu.com", shell=True)
>>>
>>> ret = subprocess.Popen("ping www.baidu.com",shell=True,stdout=subprocess.PIPE)
>>> ret.stdout.read()
创建多进程与子线程: 通过使用multiprocessing库,循环创建4个主进程,而在每个主进程内部又起了5个子线程.
import multiprocessing
import threading,os
def ThreadingFunction():
print("[-] ----> 子线程PPID: {}".format(threading.get_ident()))
def ProcessFunction(number):
print("[*] -> 主进程PID: {} 父进程: {}".format(os.getpid(),os.getppid()))
for i in range(5): # 在主进程里开辟5个线程
thread = threading.Thread(target=ThreadingFunction,) # 嵌套子线程
thread.start() # 执行子线程
if __name__ == "__main__":
for item in range(4): # 启动4个主进程
proc = multiprocessing.Process(target=ProcessFunction,args=(item,))
proc.start()
proc.join()
使用基于类的方式创建进程: 除了使用函数式方式创建进程以外,我们还可以使用基于类的方式创建.
import os,time
from multiprocessing import Process
class Myprocess(Process):
def __init__(self,person):
super().__init__()
self.person = person
def run(self):
print("[*] -> 当前PID: {}".format(os.getpid()))
print("--> 传入的人名: {}".format(self.person))
time.sleep(3)
if __name__ == '__main__':
process = Myprocess("lyshark")
#process.daemon = True # 设置p为守护进程
process.start()
进程锁(Lock): 进程中也有锁,可以实现进程之间数据的一致性,也就是进程数据的同步,保证数据不混乱.
# 由并发变成了串行,牺牲了运行效率,但避免了竞争
import multiprocessing
def func(loc,num):
loc.acquire() #添加进程锁
print("hello ---> %s" %num)
loc.release() #关闭进程锁
if __name__ == "__main__":
lock = multiprocessing.Lock() #生成进程锁
for number in range(10):
proc = multiprocessing.Process(target=func,args=(lock,number,))
proc.start()
异步进程池: 进程池内部维护一个进程序列,当使用时则去进程池中获取一个进程,如果进程池序列中没有可供使用的进程,那么程序就会等待,直到进程池中有可用进程为止.
import multiprocessing
import time
def ProcessFunction(number):
time.sleep(2)
print("[+] 进程执行ID: {}".format(number))
def ProcessCallback(arg):
print("[-] 进程执行结束,执行回调函数")
if __name__ == "__main__":
pool = multiprocessing.Pool(processes=5) # 允许进程池同时放入5个进程
for item in range(10):
pool.apply_async(func=ProcessFunction,args=(item,),callback=ProcessCallback)
pool.close()
pool.join()
from multiprocessing import Pool, TimeoutError
import time,os
def f(x):
return x*x
if __name__ == '__main__':
#启动4个工作进程作为进程池
with Pool(processes=4) as pool:
#返回函数参数运行结果列表
print(pool.map(f, range(10)))
#在进程池中以任意顺序打印相同的数字
for i in pool.imap_unordered(f, range(10)):
print(i,end=' ')
#异步评估
res = pool.apply_async(f,(20,)) #在进程池中只有一个进程运行
print('\n',res.get(timeout=1)) #打印结果，超时为1秒
#打印该进程的PID
res = pool.apply_async(os.getpid,()) #在进程池中只有一个进程运行
print(res.get(timeout=1)) #打印进程PID
#打印4个进程的PID
multiple_results = [pool.apply_async(os.getpid, ()) for i in range(4)]
print([res.get(timeout=1) for res in multiple_results])
#进程等待10秒，获取数据超时为1秒，将输出异常
res = pool.apply_async(time.sleep, (10,))
try:
print(res.get(timeout=1))
except TimeoutError:
print("We lacked patience and got a multiprocessing.TimeoutError")
8.3 多进程数据共享
一般当我们创建两个进程后,进程各自持有一份数据,默认无法共享数据,如果我们想要共享数据必须通过一个中间件来实现数据的交换,来帮你把数据进行一个投递,要实现进程之间的数据共享,其主要有以下几个方法来实现进程间数据的共享.
共享队列(Queue): 这个Queue主要实现进程与进程之间的数据共享,与线程中的Queue不同.
from multiprocessing import Process
from multiprocessing import queues
import multiprocessing
def foo(i,arg):
arg.put(i)
print('say hi',i,arg.qsize())
li = queues.Queue(20,ctx=multiprocessing)
for i in range(10):
p = Process(target=foo,args=(i,li,))
p.start()
共享整数(int): 整数之间的共享,只需要使用multiprocessing.Value方法,即可实现.
import multiprocessing
def func(num):
num.value = 1024 #虽然赋值了,但是子进程改变了这个数值
print("函数中的数值: %s"%num.value)
if __name__ == "__main__":
num = multiprocessing.Value("d",10.0) #主进程与子进程共享这个value
print("这个共享数值: %s"%num.value)
for i in range(5):
num = multiprocessing.Value("d", i) #声明进程,并传递1,2,3,4这几个数
proc = multiprocessing.Process(target=func,args=(num,))
proc.start() #启动进程
#proc.join()
print("最后打印数值: %s"%num.value)
共享数组(Array): 数组之间的共享,只需要使用multiprocessing.Array方法,即可实现.
import multiprocessing
def func(ary): #子进程改变数组,主进程跟着改变
ary[0]=100
ary[1]=200
ary[2]=300
''' i所对应的类型是ctypes.c_int,其他类型如下参考:
'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
'''
if __name__ == "__main__":
ary = multiprocessing.Array("i",[1,2,3]) #主进程与子进程共享这个数组
for i in range(5):
proc = multiprocessing.Process(target=func,args=(ary,))
print(ary[:])
proc.start()
共享字典(dict): 通过使用Manager方法,实现两个进程中的,字典与列表的数据共享.
import multiprocessing
def func(mydict, mylist):
mydict["字典1"] = "值1"
mydict["字典2"] = "值2"
mylist.append(1)
mylist.append(2)
mylist.append(3)
if __name__ == "__main__":
mydict = multiprocessing.Manager().dict() #主进程与子进程共享字典
mylist = multiprocessing.Manager().list() #主进程与子进程共享列表
proc = multiprocessing.Process(target=func,args=(mydict,mylist))
proc.start()
proc.join()
print("列表中的元素: %s" %mylist)
print("字典中的元素: %s" %mydict)
管道共享(Pipe): 通过Pipe管道的方式在两个进程之间共享数据,类似于Socket套接字.
import multiprocessing
def func(conn):
conn.send("你好我是子进程.") #发送消息给父进程
print("父进程传来了:",conn.recv()) #接收父进程传来的消息
conn.close()
if __name__ == "__main__":
parent_conn,child_conn = multiprocessing.Pipe() #管道创建两个端口,一收一发送
proc = multiprocessing.Process(target=func,args=(child_conn,))
proc.start()
print("子进程传来了:",parent_conn.recv()) #接收子进程传来的数据
parent_conn.send("我是父进程,收到消息了..") #父进程发送消息给子进程
本文作者： 王瑞
本文链接： https://www.lyshark.com/post/b4dd0803.html
版权声明： 本博客所有文章除特别声明外，均采用 BY-NC-SA 许可协议。转载请注明出处！