Python进程、线程、协程

一、概念理解

二、多进程实现

2.1、Process（用于创建进程模块）

2.1.0、方法属性

multiprocessing.Process(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)

## 参数说明
# 常用参数
target: 要要调用的函数或方法；
args/kwargs: 要传入方法的参数。args=(a,);kwargs={'name'=a,}
# 以下是不常用参数
group: 线程组，目前还没有实现，库引用中提示必须是None；
# 每个进程都属于一个进程组（PG,Process Group），进程组可以包含多个进程。进程组有一个进程组长（Leader），进程组长的ID（PID, Process ID）就作为整个进程组的ID（PGID,Process Groupd ID）。
name: 进程名；

## 方法实例
需要注意的是start()，join()，is_alive()， terminate()和exitcode方法只能由创建进程对象的过程调用。
is_alive()：返回进程是否存活。从start() 方法返回到子进程终止的那一刻，进程对象仍处于活动状态。
join([timeout])：阻塞当前上下文环境的进程程，直到调用此方法的进程终止或到达指定的timeout（可选参数）。
start()：进程准备就绪，等待CPU调度
run()：strat()调用run方法，如果实例进程时未制定传入target，这star执行t默认run()方法。
terminate()：不管任务是否完成，立即停止工作进程　
daemon：进程的守护进程标志，一个布尔值。必须在start()调用之前设置。
# 在创建线程实例的时候可以设置该参数（实例中默认为None），或者在start()前设置这个参数，如果该参数为True，则会将该进程设置为主进程的守护进程，只要是其他子进程结束且主进程执行完毕，不管被设置为守护进程的程序是否执行完，主线程都会关闭。
name：进程名字。
pid：进程号。

## 代码示例
from multiprocessing import Process
import time
import os

def info():
    print('module name:', __name__)
    print('parent process:', os.getppid())
    print('process id:', os.getpid())

def f(name):
    info()
    time.sleep(3)
    print('hello', name)

if __name__ == '__main__':
    info()
    p = Process(target=f, args=('bob',))
    # p.daemon = False
    print(p.daemon)
    p.start()
    p.join(1)
    print('name:', p.name)
    print('is_alive:', p.is_alive())
    print('exitcode:', p.exitcode)
'''
------------------------------------------------------------
module name: __main__
parent process: 1188
process id: 13060
False
module name: __mp_main__
parent process: 13060
process id: 13424
name: Process-1
is_alive: True
exitcode: None
hello bob
------------------------------------------------------------
'''

if __name__ == '__main__':
    info()
    p = Process(target=f, args=('bob',))
    p.daemon = True
    print(p.daemon)
    p.start()
    # p.join(1)
    print('name:', p.name)
    print('is_alive:', p.is_alive())
    print('exitcode:', p.exitcode)
'''
------------------------------------------------------------
module name: __main__
parent process: 1188
process id: 1668
True
name: Process-1
is_alive: True
exitcode: None
------------------------------------------------------------
'''

2.1.1、直接创建

# 1.导入包和模块
import multiprocessing
import time
 
def sing():
    for i in range(3):
        print("i am sing ooo~")
        time.sleep(0.5)
 
if __name__ == '__main__':
    # 2.使用进程类创建进程对象
    # target ：指定进程执行的函数名，不加括号
    sing_process = multiprocessing.Process(target=sing)
 
    # 3. 使用进程对象启动进程执行指定任务
    sing_process.start()

2.1.2、循环创建(🉑️；重点关注获取返回值)

# 1.导入包和模块
import multiprocessing
import time
 
def main_province2(data_prov, return_list):
		k = data_prov
		return_list.append(k)
    return return_list

return_list = multiprocessing.Manager().list()   # 使用multiprocessing的Manager下的list，dict等模块接收返回值
jobs = []
for data_prov in data_prov_list:
    p = multiprocessing.Process(target=main_province2, args=(data_prov,return_list))   # 将返回值送到函数去接收返回值
    jobs.append(p)
    p.start()   # 开始进程
for proc in jobs:
    proc.join()   # 阻塞进程，等待所有进程执行完成

return_list  # 最终的接收值

2.2、Pool（用于创建管理进程池）

2.2.0、方法属性

multiprocessing.pool.Pool([processes[, initializer[, initargs[, maxtasksperchild[, context]]]]])

## 参数详解
# processes： 是要使用的工作进程数。如果进程是None，那么使用返回的数字os.cpu_count()。也就是说根据本地的cpu个数决定，processes小于等于本地的cpu个数；
# initializer： 如果initializer是None，那么每一个工作进程在开始的时候会调用initializer(*initargs)。
# maxtasksperchild：工作进程退出之前可以完成的任务数，完成后用一个新的工作进程来替代原进程，来让闲置的资源被释放。maxtasksperchild默认是None，意味着只要Pool存在工作进程就会一直存活。
# context: 用在制定工作进程启动时的上下文，一般使用 multiprocessing.Pool() 或者一个context对象的Pool()方法来创建一个池，两种方法都适当的设置了context。

## 方法实例
apply(func [，args [，kwds ] ])
# pool.apply(test, args=(i,))
# 使用堵塞方式调用函数，必须等待上一个进程退出才能执行下一个进程
apply_async(func [，args [，kwds [，callback [，error_callback ] ] ] ])
# pool.apply_async(test, args=(i,))
# 使用非阻塞方式调用函数，可以并行执行
map(func，iterable [，chunksize ])
# pool.map(test, lists)
# 使用堵塞方式调用函数，此方法将iterable内的每一个对象作为单独的任务提交给进程池
map_async(func，iterable [，chunksize [，callback [，error_callback ] ] ])
# pool.map_async(test, range(500))
# 使用非阻塞方式调用函数，将iterable内的每一个对象作为单独的任务提交给进程池

apply_async 和 map_async 的区别


imap(func，iterable [，chunksize ])
# 使用和map相同，但是该方法将返回迭代器，使用next()获取迭代器内的内容
close()：不在接受新的任务进入进程池，等待正在运行的所有进程结束后，关闭线程池
terminal() — 立即结束所有进程，即使在运行中的进程也会被停止
join() — 主进程阻塞等待子进程执行完之后再执行， join方法要在close或terminate之后使用。（守护进程）


## 异步方法返回的结果获取
get([timeout])
# result = pool.apply_async(time.sleep, (10,))
# print(result.get(timeout=1))
# 获取返回结果，timeout可以不设置。如果timeout不是None并且结果没有在timeout秒内到达，则 multiprocessing.TimeoutError 被引发。

2.2.1、apply_async循环启用进程池中的进程

# 导入包和模块
import multiprocessing
 
# 创建进程池
pool = multiprocessing.Pool(20)
#异步调用
results = []
for data_4a in data_prov_system_acct4a_lists:
    result = pool.apply_async(main_province, args=(data_4a,))
    results.append(result)
pool.close()
pool.join()    #调用join之前，先调用close函数，否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束

#获取处理结果
data_acct4a_out_list_2 = [res.get() for res in results]
data_acct4a_out_df_2 = pd.concat(data_acct4a_out_list_2, join='inner', axis=0)

2.2.2、map_async映射启用进程池中的进程(🉑️)

# 导入包和模块
import multiproceing

# 创建进程池
pool = multiprocessing.Pool(20)
#异步调用
res = pool.map_async(main_province, data_prov_system_acct4a_lists)

#获取处理结果
data_acct4a_out_list_1 = [res_img for res_img in res.get()]
data_acct4a_out_df_1 = pd.concat(data_acct4a_out_list_1, join='inner', axis=0)

2.3、jobli包实现多线程

pip install joblib

from math import sqrt
from joblib import Parallel, delayed
Parallel(n_jobs=2)(delayed(sqrt)(i ** 2) for i in range(10))
[0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]

# Parallel(n_jobs=2): 指定两个CPU（默认是分配给不同的CPU）
# 后面的delayed(sqrt)表示要用的函数是sqrt，这里这种用法就非常类似C++里面的委托(delegate)。
# (i**2) for i in range(10): 这里注意(i**2)的括号和delayed(sqrt)是紧挨着的。这一小段表示要传递给delayed中指定的函数的参数是i^2。

2.4、队列

2.4.1、队列应用场景

2.4.2、方法属性

import multiprocessing
queue = multiprocessing.Queue(队列长度)

put	            # queue.put(数据)，放入数据(如队列已满，则程序进入阻塞状态，等待队列取出后再放入)
put_nowait	    # queue.put_nowati(数据)，放入数据(如队列已满，则不等待队列信息取出后再放入，直接报错)
get		          # queue.get(数据)，取出数据(如队列为空，则程序进入阻塞状态，等待队列防如数据后再取出)
get_nowait	    # queue.get_nowait(数据)，取出数据(如队列为空，则不等待队列放入信息后取出数据，直接报错)，放入数据后立马判断是否为空有时为True，原因是放入值和判断同时进行
qsize		        # queue.qsize()，消息数量
empty		        # queue.empty()(返回值为True或False)，判断是否为空
full	        	# queue.full()(返回值为True或False)，判断是否为满

2.4.3、实例

import multiprocessing

def square(n):
    return n * n

if __name__ == '__main__':
    data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

    # 创建进程池和队列
    pool = multiprocessing.Pool()
    queue = multiprocessing.Manager().Queue()

    # 将要处理的数据放入队列
    for n in data:
        queue.put(n)

    # 并发处理数据
    results = []
    while not queue.empty():
        n = queue.get()
        result = pool.apply_async(square, args=(n,))
        results.append(result)

    # 关闭进程池
    pool.close()
    pool.join()

    # 获取处理结果
    final_results = [r.get() for r in results]

    print(final_results)

2.5、资源查看

os.cpu_count()   # 查看cpu核数
os.getpid()      # 当前进程id
os.getppid()     # 当前父进程id

2.6、推荐阅读

三、多线程实现

python多线程详解

为什么要使用多线程？

实现多线程

import threading
from threading import Lock,Thread
import time,os


# 普通创建方式
def run(n):
    print('task',n)
    time.sleep(1)
    print('2s')
    time.sleep(1)
    print('1s')
    time.sleep(1)
    print('0s')
    time.sleep(1)

if __name__ == '__main__':
    t1 = threading.Thread(target=run,args=('t1',))     # target是要执行的函数名（不是函数），args是函数对应的参数，以元组的形式存在
    t2 = threading.Thread(target=run,args=('t2',))
    t1.start()
    t2.start()


# 自定义线程：继承threading.Thread来定义线程类，其本质是重构Thread类中的run方法
class MyThread(threading.Thread):
    def __init__(self,n):
        super(MyThread,self).__init__()   #重构run函数必须写
        self.n = n

    def run(self):
        print('task',self.n)
        time.sleep(1)
        print('2s')
        time.sleep(1)
        print('1s')
        time.sleep(1)
        print('0s')
        time.sleep(1)

if __name__ == '__main__':
    t1 = MyThread('t1')
    t2 = MyThread('t2')
    t1.start()
    t2.start()

守护线程

def run(n):
    print('task',n)
    time.sleep(1)
    print('3s')
    time.sleep(1)
    print('2s')
    time.sleep(1)
    print('1s')

if __name__ == '__main__':
    t=threading.Thread(target=run,args=('t1',))
    t.setDaemon(True)
    t.start()
    print('end')
# 通过执行结果可以看出，设置守护线程之后，当主线程结束时，子线程也将立即结束，不再执行


# 主线程等待子线程结束
# 为了让守护线程执行结束之后，主线程再结束，我们可以使用join方法，让主线程等待子线程执行
def run(n):
    print('task',n)
    time.sleep(2)
    print('5s')
    time.sleep(2)
    print('3s')
    time.sleep(2)
    print('1s')
if __name__ == '__main__':
    t=threading.Thread(target=run,args=('t1',))
    t.setDaemon(True)    #把子线程设置为守护线程，必须在start()之前设置
    t.start()
    t.join()     #设置主线程等待子线程结束
    print('end')


# 多线程共享全局变量
# 线程时进程的执行单元，进程时系统分配资源的最小执行单位，所以在同一个进程中的多线程是共享资源的
g_num = 100
def work1():
    global  g_num
    for i in range(3):
        g_num+=1
    print('in work1 g_num is : %d' % g_num)

def work2():
    global g_num
    print('in work2 g_num is : %d' % g_num)

if __name__ == '__main__':
    t1 = threading.Thread(target=work1)
    t1.start()
    time.sleep(1)
    t2=threading.Thread(target=work2)
    t2.start()

GIL和其他

'''
        由于线程之间是进行随机调度，并且每个线程可能只执行n条执行之后，当多个线程同时修改同一条数据时可能会出现脏数据，
    所以出现了线程锁，即同一时刻允许一个线程执行操作。线程锁用于锁定资源，可以定义多个锁，像下面的代码，当需要独占
    某一个资源时，任何一个锁都可以锁定这个资源，就好比你用不同的锁都可以把这个相同的门锁住一样。
        由于线程之间是进行随机调度的，如果有多个线程同时操作一个对象，如果没有很好地保护该对象，会造成程序结果的不可预期，
    我们因此也称为“线程不安全”。
        为了防止上面情况的发生，就出现了互斥锁（Lock）
'''
# def work():
#     global n
#     lock.acquire()
#     temp = n
#     time.sleep(0.1)
#     n = temp-1
#     lock.release()
#
#
# if __name__ == '__main__':
#     lock = Lock()
#     n = 100
#     l = []
#     for i in range(100):
#         p = Thread(target=work)
#         l.append(p)
#         p.start()
#     for p in l:
#         p.join()


'''
    递归锁：RLcok类的用法和Lock类一模一样，但它支持嵌套，在多个锁没有释放的时候一般会使用RLock类
'''
# def func(lock):
#     global gl_num
#     lock.acquire()
#     gl_num += 1
#     time.sleep(1)
#     print(gl_num)
#     lock.release()
#
#
# if __name__ == '__main__':
#     gl_num = 0
#     lock = threading.RLock()
#     for i in range(10):
#         t = threading.Thread(target=func,args=(lock,))
#         t.start()


'''
    信号量（BoundedSemaphore类）
    互斥锁同时只允许一个线程更改数据，而Semaphore是同时允许一定数量的线程更改数据，比如厕所有3个坑，
    那最多只允许3个人上厕所，后面的人只能等里面有人出来了才能再进去
'''
# def run(n,semaphore):
#     semaphore.acquire()   #加锁
#     time.sleep(3)
#     print('run the thread:%s\n' % n)
#     semaphore.release()    #释放
#
#
# if __name__== '__main__':
#     num=0
#     semaphore = threading.BoundedSemaphore(5)   #最多允许5个线程同时运行
#     for i in range(22):
#         t = threading.Thread(target=run,args=('t-%s' % i,semaphore))
#         t.start()
#     while threading.active_count() !=1:
#         pass
#     else:
#         print('----------all threads done-----------')

'''
    python线程的事件用于主线程控制其他线程的执行，事件是一个简单的线程同步对象，其主要提供以下的几个方法：
        clear将flag设置为 False
        set将flag设置为 True
        is_set判断是否设置了flag
        wait会一直监听flag，如果没有检测到flag就一直处于阻塞状态
    事件处理的机制：全局定义了一个Flag，当Flag的值为False，那么event.wait()就会阻塞，当flag值为True，
    那么event.wait()便不再阻塞
'''
event = threading.Event()
def lighter():
    count = 0
    event.set()         #初始者为绿灯
    while True:
        if 5 < count <=10:
            event.clear()  #红灯，清除标志位
            print("\33[41;lmred light is on...\033[0m]")
        elif count > 10:
            event.set()    #绿灯，设置标志位
            count = 0
        else:
            print('\33[42;lmgreen light is on...\033[0m')

        time.sleep(1)
        count += 1


def car(name):
    while True:
        if event.is_set():     #判断是否设置了标志位
            print('[%s] running.....'%name)
            time.sleep(1)
        else:
            print('[%s] sees red light,waiting...'%name)
            event.wait()
            print('[%s] green light is on,start going...'%name)


# startTime = time.time()
light = threading.Thread(target=lighter,)
light.start()

car = threading.Thread(target=car,args=('MINT',))
car.start()
endTime = time.time()
# print('用时：',endTime-startTime)

GIL 全局解释器

python针对不同类型的代码执行效率也是不同的

四、异步编程（协程）

异步编程的核心概念

1. 同步 vs 异步
• 同步编程：代码按顺序执行，一个任务未完成，后续任务必须等待。例如，传统的函数调用是同步的，函数未返回时，调用者会一直等待。
• 异步编程：代码可以在某些操作未完成时继续执行其他任务。等待中的操作完成时会通过回调或其他方式通知主程序。

2. 并发 vs 并行
• 并发：程序在单个处理器上通过分时机制执行多个任务，任务间交替执行，但在某一时刻只有一个任务在运行。异步编程是实现并发的一种方式。
• 并行：程序在多个处理器或多核处理器上同时执行多个任务。

3. 事件循环 (Event Loop)
• 事件循环是异步编程的调度中心。它负责管理协程的执行，并在协程暂停时切换到其他协程。asyncio 库提供了一个默认的事件循环，可以通过 asyncio.run() 来运行异步代码。事件循环会不断检查是否有准备好执行的任务，并将它们调度执行。确保在一个协程暂停时，其他协程可以利用 CPU 资源继续执行。

4. 协程 (Coroutine)
• 协程是 Python 中异步编程的基础，它是一种比线程更加轻量级的并发方式。可以在执行过程中暂停并在稍后恢复。在 Python 中，协程使用 async def 关键字定义，并通过 await 关键字来暂停执行。协程可以在等待 I/O 操作时挂起，允许其他协程继续执行。

Python 中的异步编程

关键字和函数

• async def：定义一个异步函数（协程）。

• await：等待一个异步操作完成。只能在协程中使用。(暂停当前协程，释放控制权给事件循环，等待异步操作完成。)

• asyncio.run()：启动一个异步任务。

• asyncio.create_task()：并发运行多个协程。

• asyncio.gather()：将多个协程包装成一个，等待所有协程完成。

异步编程示例

import asyncio

async def say_hello():
    print("Hello...")
    await asyncio.sleep(1)  # 模拟一个异步 I/O 操作
    print("...World!")

asyncio.run(say_hello())

• 在这个例子中，say_hello 是一个异步函数，await asyncio.sleep(1) 会让出控制权，使事件循环可以调度其他任务。

import asyncio

async def task(name, duration):
    print(f"Task {name} started...")
    await asyncio.sleep(duration)
    print(f"Task {name} finished after {duration} seconds")

async def main():
    task1 = asyncio.create_task(task("A", 2))
    task2 = asyncio.create_task(task("B", 1))

    await task1
    await task2

asyncio.run(main())

• 这里的 main() 函数通过 asyncio.create_task() 创建了两个异步任务，它们会并发执行，最终的输出顺序不会因为任务的启动顺序而改变，而是由任务的持续时间决定。

import asyncio

async def task(name, duration):
    print(f"Task {name} started...")
    await asyncio.sleep(duration)
    print(f"Task {name} finished after {duration} seconds")
    return name, duration

async def main():
    results = await asyncio.gather(
        task("A", 2),
        task("B", 1),
        task("C", 3)
    )
    print("All tasks finished:", results)

asyncio.run(main())

• asyncio.gather() 同时运行多个任务，并等待它们全部完成后再返回结果。这个示例会并发执行任务 A、B和 C，并在所有任务完成后输出结果。

import asyncio

async def task1():
    print("Task 1 started")
    await asyncio.sleep(2)  # 假设这个操作需要 2 秒
    print("Task 1 finished")

async def task2():
    print("Task 2 started")
    await asyncio.sleep(1)  # 假设这个操作需要 1 秒
    print("Task 2 finished")

async def main():
    await asyncio.gather(task1(), task2())

asyncio.run(main())

• 当 task1() 遇到 await asyncio.sleep(2) 时，任务会暂停，事件循环会切换去执行 task2()。

• task2() 遇到 await asyncio.sleep(1) 时也会暂停，事件循环又会检查是否有其他任务可以运行。

• 由于所有任务都在等待，这时事件循环会空闲等待，直到第一个 await 操作完成。

• 1 秒后，task2 的 await 操作完成，事件循环会恢复执行 task2 的后续代码，打印 Task 2 finished。

• 再过 1 秒后，task1 的 await 操作完成，事件循环会恢复执行 task1 的后续代码，打印 Task 1 finished。

异步编程的优点和注意事项

优点

1. 高效的 I/O 操作：在网络请求、文件读写等 I/O 操作中，异步编程可以显著提高效率。

2. 减少阻塞：避免程序在等待 I/O 操作时阻塞，从而更好地利用 CPU 资源。

3. 更好的响应性：在 GUI 编程或 Web 服务中，异步编程可以使程序对用户操作更加响应及时。

注意事项

1. 学习曲线：异步编程可能比同步编程更难理解，特别是在处理复杂的逻辑时。

2. 调试困难：异步代码的调试和错误处理相对复杂，需要更多的注意力。

3. 非 I/O 密集型任务：异步编程主要在 I/O 密集型任务中表现出色，而对于 CPU 密集型任务，可能并没有显著优势。

总结

五、进程、线程、协程对比

import multiprocessing
import os

def worker(num):
    print(f"Worker {num} is running on process {os.getpid()}")

if __name__ == "__main__":
    processes = []
    for i in range(5):
        p = multiprocessing.Process(target=worker, args=(i,))
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

import threading

def worker(num):
    print(f"Worker {num} is running on thread {threading.current_thread().name}")

threads = []
for i in range(5):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

import asyncio

async def worker(num):
    print(f"Worker {num} is running")
    await asyncio.sleep(1)
    print(f"Worker {num} finished")

async def main():
    tasks = [worker(i) for i in range(5)]
    await asyncio.gather(*tasks)

asyncio.run(main())

使用过程中遇到的问题

1、python版本限制进程池处理的数据量

1. 将每个进程结果写入文件，最后汇总处理，这样避免进程间大量数据传递

2. 代码中检查Python版本，低于3.8时候不使用多进程。

3. 使用dask dataframe等Python并行处理库

2、进程池中进程的数量增加，而运行效率并没有提升

3、OSError: [Errno 28] No space left on device