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("我是父进程,收到消息了..")        #父进程发送消息给子进程