Python多线程与多进程学习笔记

Posted 2023-03-03 zstar-_

本文是一篇学习笔记，学习内容主要来源于莫凡python的文档：https://mofanpy.com/tutorials/python-basic/threading/thread

多线程

线程基本结构

开启子线程的简单方式如下：

import threading

def thread_job():
    print('This is a thread of %s' % threading.current_thread())

def main():
    thread = threading.Thread(target=thread_job, )  # 定义线程
    thread.start()  # 让线程开始工作

if __name__ == '__main__':
    main()

线程阻塞

下面是一个双线程的示例，期望效果是先运行完两个子线程，再输出all done。

import threading
import time


def T1_job():
    print("T1 start\\n")
    for i in range(10):
        time.sleep(0.1)
    print("T1 finish\\n")


def T2_job():
    print("T2 start\\n")
    print("T2 finish\\n")


if __name__ == '__main__':
    thread_1 = threading.Thread(target=T1_job, name='T1')
    thread_2 = threading.Thread(target=T2_job, name='T2')
    thread_1.start()  # 开启T1
    thread_2.start()  # 开启T2
    print("all done\\n")

输出：

T1 start
T2 start
all done
T2 finish
T1 finish

实际结果发现，主线程没有“等待”子线程执行完就已经结束。
为了达到预期效果，需要通过join()方法来设定线程阻塞。

下面再开启T2之前，插入thread_1.join()

if __name__ == '__main__':
    thread_1 = threading.Thread(target=T1_job, name='T1')
    thread_2 = threading.Thread(target=T2_job, name='T2')
    thread_1.start()  # 开启T1
    thread_1.join()   
    thread_2.start()  # 开启T2
    print("all done\\n")

输出：

T1 start
T1 finish
T2 start
all done
T2 finish

可以看到，T2在等待T1结束后再开始运行。

为了达到预期情况，可以使用1221的V型排布：

if __name__ == '__main__':
    thread_1 = threading.Thread(target=T1_job, name='T1')
    thread_2 = threading.Thread(target=T2_job, name='T2')
    thread_1.start()
    thread_2.start()
    thread_2.join()
    thread_1.join()
    print("all done\\n")

输出：

T1 start
T2 start
T2 finish
T1 finish
all done

线程通信

在定义的子线程任务函数job中，无法通过return的方式将计算完成的结果返回出来。
此时，可以使用队列(Queue)这种数据结构来获取子线程的结果数据，实现线程之间的通信，下面是一个示例：

import threading
from queue import Queue


def job(l, q):
    for i in range(len(l)):
        l[i] = l[i] ** 2
    q.put(l)


def multithreading():
    q = Queue()
    threads = []
    data = [[1, 2, 3], [3, 4, 5], [4, 4, 4], [5, 5, 5]]
    for i in range(4):
        t = threading.Thread(target=job, args=(data[i], q))
        t.start()
        threads.append(t)
    for thread in threads:
        thread.join()
    results = []
    for _ in range(4):
        results.append(q.get())
    print(results)


if __name__ == '__main__':
    multithreading()

线程锁

为了防止多线程输出结果混乱，除了添加线程阻塞之外，还可以使用线程锁。
同时，线程锁还可以确保当前线程执行时，内存不会被其他线程访问，执行运算完毕后，可以打开锁共享内存。

下面是一个不添加线程锁的示例：

import threading


def job1():
    global A
    for i in range(10):
        A += 1
        print('job1', A)


def job2():
    global A
    for i in range(10):
        A += 10
        print('job2', A)


if __name__ == '__main__':
    A = 0
    t1 = threading.Thread(target=job1)
    t2 = threading.Thread(target=job2)
    t1.start()
    t2.start()
    t1.join()
    t2.join()

输出：

job1 1
job1 2
job1 3
job1 4
job1 5
job1 6
job1 7
job1 8
job2job1 19
18
job2 29
job2 39job1
job2 50
job240
60
job2 70
job2 80
job2 90
job2 100
job2 110

添加线程锁之后：

import threading


def job1():
    global A, lock
    lock.acquire()
    for i in range(10):
        A += 1
        print('job1', A)
    lock.release()


def job2():
    global A, lock
    lock.acquire()
    for i in range(10):
        A += 10
        print('job2', A)
    lock.release()


if __name__ == '__main__':
    lock = threading.Lock()
    A = 0
    t1 = threading.Thread(target=job1)
    t2 = threading.Thread(target=job2)
    t1.start()
    t2.start()
    t1.join()
    t2.join()

输出：

job1 1
job1 2
job1 3
job1 4
job1 5
job1 6
job1 7
job1 8
job1 9
job1 10
job2 20
job2 30
job2 40
job2 50
job2 60
job2 70
job2 80
job2 90
job2 100
job2 110

多进程

进程基本结构

import multiprocessing as mp

def job(a, d):
    print('aaaaa')

if __name__ == '__main__':
    p1 = mp.Process(target=job, args=(1, 2))
    p1.start()
    p1.join()

进程通信

和多线程类似，进程之间同样可以通过队列queue形式进行通信，并且，在multiprocessing库中，直接包含了Queue()结构。

import multiprocessing as mp


def job(q):
    res = 0
    for i in range(1000):
        res += i + i ** 2 + i ** 3
    q.put(res)  # queue


if __name__ == '__main__':
    q = mp.Queue()
    p1 = mp.Process(target=job, args=(q,))
    p2 = mp.Process(target=job, args=(q,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()
    res1 = q.get()
    res2 = q.get()
    print(res1 + res2)

进程池

进程池(Pool)就是将所要运行的东西，放到池子里，Python会自行解决多进程的问题。

下面是一个简单示例，processes参数指定进程池中的进程数。

import multiprocessing as mp


def job(x):
    return x * x


def multicore():
    pool = mp.Pool(processes=6)
    # res = pool.map(job, range(10))
    # print(res)
    res = pool.apply_async(job, (2,))
    print(res.get())


if __name__ == '__main__':
    multicore()

进程池运算结果有两种获取方式：

• 第一种是pool.map，在map()中需要放入函数和需要迭代运算的值，然后它会自动分配给CPU核，返回结果；
• 第二种是pool.apply_async()，在apply_async()中只能传递一个值，它只会放入一个核进行运算，但是传入值时要注意是可迭代的，所以在传入值后需要加逗号, 同时需要用get()方法获取返回值。

pool.apply_async()只能传递一个值，如果要传递多个值，可以使用迭代器，下面的代码通过迭代器实现了两种取值方式的等效结果：

import multiprocessing as mp


def job(x):
    return x * x


def multicore():
    pool = mp.Pool(processes=2)
    res = pool.map(job, range(10))
    print(res)
    res = pool.apply_async(job, (2,))
    # 用get获得结果
    print(res.get())
    # 迭代器，i=0时apply一次，i=1时apply一次等等
    multi_res = [pool.apply_async(job, (i,)) for i in range(10)]
    # 从迭代器中取出
    print([res.get() for res in multi_res])


if __name__ == '__main__':
    multicore()

进程锁

在不同进程中，可以通过变量.value的方式共享变量内存。
如果多个进程对同一个变量进行操控，不加进程锁，就会让结果混乱。
下面是一个不加进程锁的示例：

import multiprocessing as mp
import time


def job(v, num):
    for _ in range(5):
        time.sleep(0.1)  # 暂停0.1秒，让输出效果更明显
        v.value += num  # v.value获取共享变量值
        print(v.value)


def multicore():
    v = mp.Value('i', 0)  # 定义共享变量
    p1 = mp.Process(target=job, args=(v, 1))
    p2 = mp.Process(target=job, args=(v, 3))  # 设定不同的number看如何抢夺内存
    p1.start()
    p2.start()
    p1.join()
    p2.join()


if __name__ == '__main__':
    multicore()

输出：

1
2
3
4
7
8
11
14
17
20

添加进程锁之后：

import multiprocessing as mp
import time


def job(v, num, l):
    l.acquire()  # 锁住
    for _ in range(5):
        time.sleep(0.1)
        v.value += num  # 获取共享内存
        print(v.value)
    l.release()  # 释放


def multicore():
    l = mp.Lock()  # 定义一个进程锁
    v = mp.Value('i', 0)  # 定义共享内存
    p1 = mp.Process(target=job, args=(v, 1, l))  # 需要将lock传入
    p2 = mp.Process(target=job, args=(v, 3, l))
    p1.start()
    p2.start()
    p1.join()
    p2.join()


if __name__ == '__main__':
    multicore()

输出：

1
2
3
4
5
8
11
14
17
20

进程锁保证了进程p1的完整运行，然后才进行了进程p2的运行。

多线程和多进程的效率对比

在python语言中，并无法做到实际的多线程，这是由于Python中内置了全局解释器锁(GIL)，让任何时候只有一个线程进行执行。下面是一段具体解释：

尽管Python完全支持多线程编程， 但是解释器的C语言实现部分在完全并行执行时并不是线程安全的。 实际上，解释器被一个全局解释器锁保护着，它确保任何时候都只有一个Python线程执行。 GIL最大的问题就是Python的多线程程序并不能利用多核CPU的优势 （比如一个使用了多个线程的计算密集型程序只会在一个单CPU上面运行）。
在讨论普通的GIL之前，有一点要强调的是GIL只会影响到那些严重依赖CPU的程序（比如计算型的）。 如果你的程序大部分只会涉及到I/O，比如网络交互，那么使用多线程就很合适， 因为它们大部分时间都在等待。实际上，你完全可以放心的创建几千个Python线程， 现代操作系统运行这么多线程没有任何压力，没啥可担心的。

下面是一段测试程序，对比常规计算，双线程，双进程的计算效率：

import multiprocessing as mp
import threading as td
import time


def job(q):
    res = 0
    for i in range(1000000):
        res += i + i ** 2 + i ** 3
    q.put(res)  # queue


def multicore():
    q = mp.Queue()
    p1 = mp.Process(target=job, args=(q,))
    p2 = mp.Process(target=job, args=(q,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()
    res1 = q.get()
    res2 = q.get()
    print('multicore:', res1 + res2)


def multithread():
    q = mp.Queue()  # thread可放入process同样的queue中
    t1 = td.Thread(target=job, args=(q,))
    t2 = td.Thread(target=job, args=(q,))
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    res1 = q.get()
    res2 = q.get()
    print('multithread:', res1 + res2)


def normal():
    res = 0
    for _ in range(2):
        for i in range(1000000):
            res += i + i ** 2 + i ** 3
    print('normal:', res)


if __name__ == '__main__':
    st = time.time()
    normal()
    st1 = time.time()
    print('normal time:', st1 - st)
    multithread()
    st2 = time.time()
    print('multithread time:', st2 - st1)
    multicore()
    print('multicore time:', time.time() - st2)

输出：

normal: 499999666667166666000000
normal time: 0.9803786277770996
multithread: 499999666667166666000000
multithread time: 0.9883582592010498
multicore: 499999666667166666000000
multicore time: 1.4371891021728516

结果发现，双线程的所花时间和单线程相差不大，论证了python的多线程是“伪多线程”。然而，多进程的所花时间却更多，这是由于该运算较简单，启动线程的时间消耗过大。

把计算数扩大十倍，输出结果：

normal: 4999999666666716666660000000
normal time: 10.019219160079956
multithread: 4999999666666716666660000000
multithread time: 9.802824974060059
multicore: 4999999666666716666660000000
multicore time: 6.478690147399902

此时发现多进程的速度有了明显提升。