8.0 Python 使用进程与线程

十年风暴

python 进程与线程是并发编程的两种常见方式。进程是操作系统中的一个基本概念，表示程序在操作系统中的一次执行过程，拥有独立的地址空间、资源、优先级等属性。线程是进程中的一条执行路径，可以看做是轻量级的进程，与同一个进程中的其他线程共享相同的地址空间和资源。
线程和进程都可以实现并发编程，但是它们之间有几点不同：

• 线程间共享进程的内存空间，但进程间的内存空间是相互独立的；
  
• 线程创建和销毁的开销较小，但是线程切换的开销较大；
  
• 进程间通信需要较为复杂的 IPC（Inter-Process Communication）机制，线程间通信则可以直接读写共享内存；
  
• 多进程可以充分利用多核 CPU 的性能，但是多线程受 GIL（Global Interpreter Lock）限制，只能利用单核 CPU 的性能。
  

在选择使用进程还是线程时，需要根据具体场景和需求进行权衡和选择。如果任务需要充分利用多核 CPU，且任务之间互不影响，可以选择多进程；如果任务之间需要共享资源和数据，可以选择多线程。同时，需要注意在 python 中使用多线程时，由于 GIL 的存在，可能无法实现真正的并行。
8.1 创建并使用线程

线程是操作系统调度的最小执行单元，是进程中的一部分，能够提高程序的效率。在python中，创建线程需要使用threading模块。该模块的实现方法是底层调用了C语言的原生函数来实现线程的创建和管理。在系统中，所有的线程看起来都是同时执行的，但实际上是由操作系统进行时间片轮转调度的。
使用函数创建线程: 创建线程并传递参数实现指定函数多线程并发,使用join方法,等待线程执行完毕后的返回结果.

1. import os,time
2. import threading
3. now = lambda:time.time()
5. def MyThread(x,y):                         # 定义每个线程要执行的函数体
6.     time.sleep(5)                          # 睡眠5秒钟
7.     print("传递的数据:%s,%s"%(x,y))         # 其中有两个参数,我们动态传入
9. if __name__ == "__main__":
10.     ThreadPool = []
12.     start = now()
13.     for item in range(10):                                             # 创建10个线程并发执行函数
14.         thread = threading.Thread(target=MyThread,args=(item,item+1,)) # args =>函数的参数
15.         thread.start()                                                 # 启动线程
16.         ThreadPool.append(thread)
17.     for item in ThreadPool:
18.         item.join()
19.         print("[+] 线程信息: {}".format(item))
20.     stop = now()
21.     print("[+] 线程总耗时: {} s".format(int(stop-start)))

复制代码

使用类创建内部线程: 通过定义类,将线程函数与类进行结合实现一体化该方式调用方便思维明确.

1. import os,time
2. import threading
4. class MyThread(threading.Thread):
5.     def __init__(self,x,y):
6.         super(MyThread, self).__init__()
7.         self.x = x
8.         self.y = y
10.     def run(self):                       # 用于执行相应操作(固定写法)
11.         print("[+] 当前执行运算: {} + {}".format(self.x,self.y))
12.         self.result = self.x + self.y
14.     def get_result(self):                # 获取计算结果
15.         try:
16.             return self.result
17.         except Exception:
18.             return None
20. if __name__ == "__main__":
21.     ThreadPool = []
23.     for item in range(1,10):
24.         obj = MyThread(item,item+1)
25.         obj.start()
26.         ThreadPool.append(obj)
28.     for item in ThreadPool:
29.         item.join()
30.         print("[+] 获取返回: ",item.get_result())

复制代码

使用类创建外部线程: 该定义方式与上方完全不同,我们可以将执行过程定义到类的外部为单独函数,然后类内部去调用传参.

1. import os,time
2. import threading
4. def MyThreadPrint(x,y):
5.     print("[+] 当前执行运算: {} + {}".format(x,y))
6.     result = x + y
7.     return result
9. class MyThread(threading.Thread):
10.     def __init__(self,func,args=()):
11.         super(MyThread, self).__init__()
12.         self.func = func
13.         self.args = args
15.     def run(self):
16.         self.result = self.func(*self.args)
18.     def get_result(self):
19.         try:
20.             return self.result
21.         except Exception:
22.             return None
24. if __name__ == "__main__":
25.     ThreadPool = []
27.     for item in range(1,10):
28.         obj = MyThread(func=MyThreadPrint,args=(item,item+1))
29.         obj.start()
30.         ThreadPool.append(obj)
32.     for item in ThreadPool:
33.         item.join()
34.         print("[+] 获取返回: ",item.get_result())

复制代码

在线程中创建子线程: 通过创建一个守护线程,并让守护线程调用子线程,从而实现线程中调用线程,线程嵌套调用.

1. import time
2. import threading
4. # run => 子线程 => 由主线程调用它
5. def run(num):
6.     print("这是第 {} 个子线程".format(num))
7.     time.sleep(2)
9. # main = > 主守护线程 => 在里面运行5个子线程
10. def main():
11.     for each in range(5):
12.         thread = threading.Thread(target=run,args=(each,))
13.         thread.start()
14.         print("启动子线程: {} 编号: {}".format(thread.getName(),each))
15.     thread.join()
17. if __name__ == "__main__":
18.     daemon = threading.Thread(target=main,args=())
19.     daemon.setDaemon(True)   # 设置主线程为守护线程
20.     daemon.start()           # 启动守护线程
21.     daemon.join(timeout=10)  # 设置10秒后关闭,不论子线程是否执行完毕

复制代码

简单的线程互斥锁(Semaphore): 同时允许一定数量的线程更改数据,也就是限制每次允许执行的线程数.

1. import threading,time
2. semaphore = threading.BoundedSemaphore(5)         #最多允许5个线程同时运行
4. def run(n):
5.     semaphore.acquire()                           #添加信号
6.     time.sleep(1)
7.     print("运行这个线程中: %s"%n)
8.     semaphore.release()                           #关闭信号
10. if __name__ == '__main__':
11.     for i in range(20):                           #同时执行20个线程
12.         t = threading.Thread(target=run, args=(i,))
13.         t.start()
15. while threading.active_count() != 1:              #等待所有线程执行完毕
16.     pass
17. else:
18.     print('----所有线程执行完毕了---')

复制代码

1. import threading,time
3. class mythreading(threading.Thread):
4.     def run(self):
5.         semaphore.acquire()  #获取信号量锁
6.         print('running the thread:',self.getName())
7.         time.sleep(2)
8.         semaphore.release()  #释放信号量锁
10. if __name__ == "__main__":
11.     semaphore = threading.BoundedSemaphore(3) # 只运行3个线程同时运行
12.     for i in range(20):
13.         t1 = mythreading()
14.         t1.start()
15.     t1.join()

复制代码

线程全局锁(Lock): 添加本全局锁以后,能够保证在同一时间内保证只有一个线程具有权限.

1. import time
2. import threading
4. num = 0                  #定义全局共享变量
5. thread_list = []         #线程列表
6. lock = threading.Lock()  #生成全局锁
8. def SumNumber():
9.     global num          #在每个线程中获取这个全局变量
10.     time.sleep(2)
11.     lock.acquire()      #修改数据前给数据加锁
12.     num += 1            #每次进行递增操作
13.     lock.release()      #执行完毕以后,解除锁定
16. for x in range(50):     #指定生成线程数
17.     thread = threading.Thread(target=SumNumber)
18.     thread.start()              #启动线程
19.     thread_list.append(thread)  #将结果列表加入到变量中
21. for y in thread_list:           #等待执行完毕.
22.     y.join()
24. print("计算结果: ",num)

复制代码

线程递归锁(RLock): 递归锁和全局锁差不多,递归锁就是在大锁中还要添加个小锁,递归锁是常用的锁.

1. import threading
2. import time
4. num = 0                          #初始化全局变量
5. lock = threading.RLock()         #设置递归锁
7. def fun1():
8.     lock.acquire()              #添加递归锁
9.     global num
10.     num += 1
11.     lock.release()              #关闭递归锁
12.     return num
14. def fun2():
15.     lock.acquire()              #添加递归锁
16.     res = fun1()
17.     print("计算结果: ",res)
18.     lock.release()              #关闭递归锁
20. if __name__ == "__main__":
21.     for x in range(10):         #生成10个线程
22.         thread = threading.Thread(target=fun2)
23.         thread.start()
25. while threading.active_count() != 1:   #等待所有线程执行完成
26.     print(threading.active_count())
27. else:
28.     print("所有线程运行完成...")
29.     print(num)

复制代码

线程互斥锁量控制并发: 使用BoundedSemaphore定义默认信号10,既可以实现控制单位时间内的程序并发量.

1. import os,time
2. import threading
4. def MyThread(x):
5.     lock.acquire()       # 上锁
6.     print("执行数据: {}".format(x))
7.     lock.release()       # 释放锁
8.     time.sleep(2)        # 模拟函数消耗时间
9.     threadmax.release()  # 释放信号,可用信号加1
11. if __name__ == "__main__":
12.     # 此处的BoundedSemaphore就是说默认给与10个信号
13.     threadmax = threading.BoundedSemaphore(10)  # 限制线程的最大数量为10个
14.     lock = threading.Lock()   # 将锁内的代码串行化(防止print输出乱行)
15.     ThreadPool = []           # 执行线程池
17.     for item in range(1,100):
18.         threadmax.acquire()  # 增加信号,可用信号减1
19.         thread = threading.Thread(target=MyThread,args=(item,))
20.         thread.start()
21.         ThreadPool.append(thread)
23.     for item in ThreadPool:
24.         item.join()

复制代码

线程驱动事件(Event): 线程事件用于主线程控制其他线程的执行,事件主要提供了三个方法set、wait、clear、is_set,分别用于设置检测和清除标志.

1. import threading
2. event = threading.Event()
4. def func(x,event):
5.     print("函数被执行了: %s 次.." %x)
6.     event.wait()         # 检测标志位状态,如果为True=继续执行以下代码,反之等待.
7.     print("加载执行结果: %s" %x)
9. for i in range(10):      # 创建10个线程
10.     thread = threading.Thread(target=func,args=(i,event,))
11.     thread.start()
13. print("当前状态: %s" %event.is_set())      # 检测当前状态,这里为False
14. event.clear()                             # 将标志位设置为False,默认为False
15. temp=input("输入yes解锁新姿势: ")          # 输入yes手动设置为True
16. if temp == "yes":
17.     event.set()                           # 设置成True
18.     print("当前状态: %s" %event.is_set())  # 检测当前状态,这里为True

复制代码

1. import threading
3. def show(event):
4.     event.wait()                     # 阻塞线程执行程序
5.     print("执行一次线程操作")
7. if __name__ == "__main__":
8.     event_obj = threading.Event()    # 创建event事件对象
9.     for i in range(10):
10.         t1 = threading.Thread(target=show,args=(event_obj,))
11.         t1.start()
12.         inside = input('>>>:')
13.         if inside == '1':
14.             event_obj.set() # 当用户输入1时set全局Flag为True,线程不再阻塞
15.         event_obj.clear()   # 将Flag设置为False

复制代码

线程实现条件锁: 条件(Condition) 使得线程等待,只有满足某条件时,才释放N个线程.

1. import threading
3. def condition_func():
4.     ret = False
5.     inp = input(">> ")
6.     if inp == '1':
7.         ret = True
8.     return ret
10. def run(n):
11.     con.acquire()                # 条件锁
12.     con.wait_for(condition_func) # 判断条件
13.     print('running...',n)
14.     con.release()                # 释放锁
16. if __name__ == "__main__":
17.     con = threading.Condition()  # 建立线程条件对象
18.     for i in range(10):
19.         t = threading.Thread(target=run,args=(i,))
20.         t.start()
21.         t.join()

复制代码

单线程异步并发执行: 在单线程下实现异步执行多个函数,返回耗时取决于最后一个函数的执行时间.

1. import time,asyncio
3. now = lambda :time.time()
5. async def GetSystemMem(sleep):
6.     print("[+] 执行获取内存异步函数.")
7.     await asyncio.sleep(sleep)   # 设置等待时间
8.     return 1
10. async def GetSystemCPU(sleep):
11.     print("[+] 执行获取CPU异步函数.")
12.     await asyncio.sleep(sleep)   # 设置等待时间
13.     return 1
15. if __name__ == "__main__":
16.     stop = now()
17.     mem = GetSystemMem(1)
18.     cpu = GetSystemCPU(4)
20.     task=[
21.         asyncio.ensure_future(mem),             # 将多个任务添加进一个列表
22.         asyncio.ensure_future(cpu)
23.     ]
24.     loop=asyncio.get_event_loop()               # 创建一个事件循环
25.     loop.run_until_complete(asyncio.wait(task)) # 开始并发执行
27.     for item in task:
28.         print("[+] 返回结果: ",item.result())    # 输出回调
29.     print('总耗时: {}'.format(stop - now()))

复制代码

8.2 创建并使用进程

进程是指正在执行的程序，创建进程需要使用multiprocessing模块，创建方法和线程相同，但由于进程之间的数据需要各自持有一份，所以创建进程需要更大的开销。进程间数据不共享，多进程可以用来处理多任务，但很消耗资源。计算密集型任务最好交给多进程来处理，I/O密集型任务最好交给多线程来处理。另外，进程的数量应该和CPU的核心数保持一致，以充分利用系统资源。
使用进程函数执行命令: 通过系统提供的进程线程函数完成对系统命令的调用与执行.

1. >>> import os,subprocess
2. >>>
3. >>> os.system("ping -n 1 www.baidu.com")       # 在当前shell中执行命令
4. >>>
5. >>> ret = os.popen("ping -n 1 www.baidu.com")  # 在子shell中执行命令
6. >>> ret.read()
7. >>>
8. >>> subprocess.run("ping www.baidu.com",shell=True)
9. >>> subprocess.call("ping www.baidu.com", shell=True)
10. >>>
11. >>> ret = subprocess.Popen("ping www.baidu.com",shell=True,stdout=subprocess.PIPE)
12. >>> ret.stdout.read()

复制代码

创建多进程与子线程: 通过使用multiprocessing库,循环创建4个主进程,而在每个主进程内部又起了5个子线程.

1. import multiprocessing
2. import threading,os
4. def ThreadingFunction():
5.     print("[-] ----> 子线程PPID: {}".format(threading.get_ident()))
7. def ProcessFunction(number):
8.     print("[*] -> 主进程PID: {} 父进程: {}".format(os.getpid(),os.getppid()))
9.     for i in range(5):                                       # 在主进程里开辟5个线程
10.         thread = threading.Thread(target=ThreadingFunction,) # 嵌套子线程
11.         thread.start()                                       # 执行子线程
13. if __name__ == "__main__":
14.     for item in range(4):                                    # 启动4个主进程
15.         proc = multiprocessing.Process(target=ProcessFunction,args=(item,))
16.         proc.start()
17.         proc.join()

复制代码

使用基于类的方式创建进程: 除了使用函数式方式创建进程以外,我们还可以使用基于类的方式创建.

1. import os,time
2. from multiprocessing import Process
4. class Myprocess(Process):
5.     def __init__(self,person):
6.         super().__init__()
7.         self.person = person
9.     def run(self):
10.         print("[*] -> 当前PID: {}".format(os.getpid()))
11.         print("--> 传入的人名: {}".format(self.person))
12.         time.sleep(3)
14. if __name__ == '__main__':
15.     process = Myprocess("lyshark")
16.     #process.daemon = True # 设置p为守护进程
17.     process.start()

复制代码

进程锁(Lock): 进程中也有锁,可以实现进程之间数据的一致性,也就是进程数据的同步,保证数据不混乱.

1. # 由并发变成了串行,牺牲了运行效率,但避免了竞争
2. import multiprocessing
4. def func(loc,num):
5.     loc.acquire()                        #添加进程锁
6.     print("hello ---> %s" %num)
7.     loc.release()                        #关闭进程锁
9. if __name__ == "__main__":
10.     lock = multiprocessing.Lock()        #生成进程锁
12.     for number in range(10):
13.         proc = multiprocessing.Process(target=func,args=(lock,number,))
14.         proc.start()

复制代码

异步进程池: 进程池内部维护一个进程序列,当使用时则去进程池中获取一个进程,如果进程池序列中没有可供使用的进程,那么程序就会等待,直到进程池中有可用进程为止.

1. import multiprocessing
2. import time
4. def ProcessFunction(number):
5.     time.sleep(2)
6.     print("[+] 进程执行ID: {}".format(number))
8. def ProcessCallback(arg):
9.     print("[-] 进程执行结束,执行回调函数")
11. if __name__ == "__main__":
12.     pool = multiprocessing.Pool(processes=5)               # 允许进程池同时放入5个进程
13.     for item in range(10):
14.         pool.apply_async(func=ProcessFunction,args=(item,),callback=ProcessCallback)
15.     pool.close()
16.     pool.join()

复制代码

1. from multiprocessing import Pool, TimeoutError
2. import time,os
4. def f(x):
5.     return x*x
7. if __name__ == '__main__':
8.     #启动4个工作进程作为进程池
9.     with Pool(processes=4) as pool:
10.         #返回函数参数运行结果列表
11.         print(pool.map(f, range(10)))
12.         #在进程池中以任意顺序打印相同的数字
13.         for i in pool.imap_unordered(f, range(10)):
14.             print(i,end=' ')
15.         #异步评估
16.         res = pool.apply_async(f,(20,))      #在进程池中只有一个进程运行
17.         print('\n',res.get(timeout=1))       #打印结果，超时为1秒
18.         #打印该进程的PID
19.         res = pool.apply_async(os.getpid,()) #在进程池中只有一个进程运行
20.         print(res.get(timeout=1))            #打印进程PID
22.         #打印4个进程的PID
23.         multiple_results = [pool.apply_async(os.getpid, ()) for i in range(4)]
24.         print([res.get(timeout=1) for res in multiple_results])
26.         #进程等待10秒，获取数据超时为1秒，将输出异常
27.         res = pool.apply_async(time.sleep, (10,))
28.         try:
29.             print(res.get(timeout=1))
30.         except TimeoutError:
31.             print("We lacked patience and got a multiprocessing.TimeoutError")

复制代码

8.3 多进程数据共享

一般当我们创建两个进程后,进程各自持有一份数据,默认无法共享数据,如果我们想要共享数据必须通过一个中间件来实现数据的交换,来帮你把数据进行一个投递,要实现进程之间的数据共享,其主要有以下几个方法来实现进程间数据的共享.
共享队列(Queue): 这个Queue主要实现进程与进程之间的数据共享,与线程中的Queue不同.

1. from multiprocessing import Process
2. from multiprocessing import queues
3. import multiprocessing
5. def foo(i,arg):
6.     arg.put(i)
7.     print('say hi',i,arg.qsize())
9. li = queues.Queue(20,ctx=multiprocessing)
11. for i in range(10):
12.     p = Process(target=foo,args=(i,li,))
13.     p.start()

复制代码

共享整数(int): 整数之间的共享,只需要使用multiprocessing.Value方法,即可实现.

1. import multiprocessing
3. def func(num):
4.     num.value = 1024                              #虽然赋值了,但是子进程改变了这个数值
5.     print("函数中的数值: %s"%num.value)
8. if __name__ == "__main__":
9.     num = multiprocessing.Value("d",10.0)         #主进程与子进程共享这个value
10.     print("这个共享数值: %s"%num.value)
12.     for i in range(5):
13.         num = multiprocessing.Value("d", i)      #声明进程,并传递1,2,3,4这几个数
14.         proc = multiprocessing.Process(target=func,args=(num,))
15.         proc.start()                             #启动进程
16.         #proc.join()
17.         print("最后打印数值: %s"%num.value)

复制代码

共享数组(Array): 数组之间的共享,只需要使用multiprocessing.Array方法,即可实现.

1. import multiprocessing
4. def func(ary):       #子进程改变数组,主进程跟着改变
5.     ary[0]=100
6.     ary[1]=200
7.     ary[2]=300
9. ''' i所对应的类型是ctypes.c_int,其他类型如下参考:
10.     'c': ctypes.c_char,  'u': ctypes.c_wchar,
11.     'b': ctypes.c_byte,  'B': ctypes.c_ubyte,
12.     'h': ctypes.c_short, 'H': ctypes.c_ushort,
13.     'i': ctypes.c_int,   'I': ctypes.c_uint,
14.     'l': ctypes.c_long,  'L': ctypes.c_ulong,
15.     'f': ctypes.c_float, 'd': ctypes.c_double
16. '''
18. if __name__ == "__main__":
19.     ary = multiprocessing.Array("i",[1,2,3])   #主进程与子进程共享这个数组
21.     for i in range(5):
22.         proc = multiprocessing.Process(target=func,args=(ary,))
23.         print(ary[:])
24.         proc.start()

复制代码

共享字典(dict): 通过使用Manager方法,实现两个进程中的,字典与列表的数据共享.

1. import multiprocessing
3. def func(mydict, mylist):
4.     mydict["字典1"] = "值1"
5.     mydict["字典2"] = "值2"
6.     mylist.append(1)
7.     mylist.append(2)
8.     mylist.append(3)
10. if __name__ == "__main__":
12.     mydict = multiprocessing.Manager().dict()        #主进程与子进程共享字典
13.     mylist = multiprocessing.Manager().list()        #主进程与子进程共享列表
15.     proc = multiprocessing.Process(target=func,args=(mydict,mylist))
16.     proc.start()
17.     proc.join()
19.     print("列表中的元素: %s" %mylist)
20.     print("字典中的元素: %s" %mydict)

复制代码

管道共享(Pipe): 通过Pipe管道的方式在两个进程之间共享数据,类似于Socket套接字.

1. import multiprocessing
3. def func(conn):
4.     conn.send("你好我是子进程.")                      #发送消息给父进程
5.     print("父进程传来了:",conn.recv())                #接收父进程传来的消息
6.     conn.close()
8. if __name__ == "__main__":
9.     parent_conn,child_conn = multiprocessing.Pipe()  #管道创建两个端口,一收一发送
10.     proc = multiprocessing.Process(target=func,args=(child_conn,))
11.     proc.start()
13.     print("子进程传来了:",parent_conn.recv())         #接收子进程传来的数据
14.     parent_conn.send("我是父进程,收到消息了..")        #父进程发送消息给子进程

复制代码

本文作者： 王瑞
本文链接： https://www.lyshark.com/post/b4dd0803.html
版权声明： 本博客所有文章除特别声明外，均采用 BY-NC-SA 许可协议。转载请注明出处！

来源:https://www.cnblogs.com/LyShark/p/17628767.html
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容，谢谢合作！