提供了Python编程的基础技术教程

Python程序教程

您现在的位置是：首页 >  python编程

当前栏目

Python 编程 | 连载 26 - Python 多线程

Python,编程,连载,26,多线程
2025-03-13 21:27:22 时间

持续创作，加速成长！这是我参与「掘金日新计划 · 6 月更文挑战」的第1天，点击查看活动详情

一、多线程

线程是系统的最小调度单元，线程相比进程来说，对于资源的消耗低。线程可以通过threading模块下Thread函数来创建，线程对象的相关方法有：

  • Thread：创建线程，入参需要传入函数名以及函数的参数，返回一个线程对象
  • start：启动线程
  • join：阻塞直到线程执行结束
  • getName：获取线程名
  • setName：设置线程名
  • is_alive：判断线程是否存活
  • setDaemon：守护线程

通过random.choice函数选中一个列中的元素，从列表中移除该元素并加入另外一个列表，直至列表为空。

import random, time

heros = ['stark', 'clint', 'thor', 'hulk', 'widow', 'captain', 'park', 'loki', 'strange', 'wanda']
_heros = []


def create():
    if len(heros) == 0:
        return
    hero = random.choice(heros)
    heros.remove(hero)
    print('移除的元素为：{}'.format(hero))
    _heros.append(hero)
    time.sleep(1)


if __name__ == '__main__':
    start = time.time()
    for i in range(len(heros)):
        create()
    print('heros: {}'.format(heros))
    print('_heros: {}'.format(_heros))
    end = time.time()
    print('耗时: {}'.format(end - start))

使用多线程方式处理，首先导入threading模块，修改main函数中的代码。

if __name__ == '__main__':
    start = time.time()
    threads = []
    for i in range(len(heros)):
        # 创建线程执行任务
        thread = threading.Thread(target=create)
        threads.append(thread)
        thread.start()
        print('线程名为：{}'.format(thread.getName()))

    for thread in threads:
        thread.join()
        # create()
    print('heros: {}'.format(heros))
    print('_heros: {}'.format(_heros))
    end = time.time()
    print('耗时: {}'.format(end - start))

循环创建了10个线程，每个线程都去执行任务，整个耗时非常短。

通过线程执行任务存在的问题：

  • 函数无法获取返回值
  • 多个线程同时修改文件可能造成数据混乱
  • 线程太多可能会造成资源不足

二、线程之间的通信

线程之间通信同样需要使用到队列。

import threading, time, queue, random

def send():
    while True:
        mes = random.randint(1, 10)
        print('send message {}'.format(mes))
        queue.put(mes)
        time.sleep(1)


def receive():
    while True:
        x = queue.get()
        print('receive {}'.format(x))
        time.sleep(1)


if __name__ == '__main__':
    queue = queue.Queue()
    t1 = threading.Thread(target=send)
    t2 = threading.Thread(target=receive)

    t1.start()
    t2.start()

    t1.join()
    t2.join()

线程池的创建

线程池可以避免线程创建和销毁带来的消耗，线程池需要通过futures.ThreadPoolExecutir方法来创建，线程池相关的方法有

  • futures.ThreadPoolExecutor：创建线程池
  • submit：往线程池中加入任务
  • done：判断线程池中的某个线程是否完成任务
  • result：获取线程执行的结果

首先导入concurrent.futures.thread包

import time, threading
from concurrent.futures.thread import ThreadPoolExecutor


def hallo(info):
    print(info)
    time.sleep(1)


if __name__ == '__main__':
    # 创建线程池
    pool = ThreadPoolExecutor(2)
    for i in range(20):
        pool.submit(hallo, ('Mark {}'.format(i)))

控制台输出结果时几乎是两条信息同时打印，这是因为此时有两个线程正在执行任务。

# 创建一个线程锁
lock = threading.Lock()

def hallo(info):
    # 上锁
    lock.acquire()
    print(info)
    time.sleep(1)
    # 开锁
    lock.release()

上了线程锁，就只能一个一个的执行了。

修改hallo()函数，返回info参数，并将上锁和解锁代码注释

# 其余代码不变
def hallo(info):
    # 上锁
    # lock.acquire()
    # print(info)
    time.sleep(1)
    # 开锁
    # lock.release()
    # print('PID：{}'.format(os.getpid()))
    return info


if __name__ == '__main__':
    # hallo()函数返回结果的列表
    results = []
    # 创建线程池
    pool = ThreadPoolExecutor(2)
    for i in range(20):
        res = pool.submit(hallo, ('Mark {}'.format(i)))
        results.append(res)
        
    print(res)
    print(dir(res))
    print('是否执行结束：{}'.format(res.done()))


    # 遍历结果列表
    for res in results:
        print('遍历结果列表：{}'.format(res.result()))
    print('是否执行结束：{}'.format(res.done()))

线程池通过submit提交一个任务执行，返回一个Future对象，可以从该对象中通过调用result()函数获取任务执行的返回值。

GIL全局锁

Python 解释器在执行的时候自动加的一把锁，造成Python中的多线程无法在多个core执行，只能在一个core上执行，这把锁就是GIL锁。GIL是全局解释器锁，并不是Python的特性，它是在Cpython解释器里引入的一个概念，而在其他语言编写的解释器里没有GIL。

GIL锁的作用：

  • 单一CPU工作
  • 确保线程安全

pypy解释器是没有GIL全局锁的，但是不推荐使用pypy解释器，推荐多进程+多线程的方式，通过多个进程在多个CPU上执行，每个进程在执行多个线程。

在CPython解释其中，当Python代码有一个线程开始访问解释器的时候，GIL就会给这个线程上锁，此时此刻线程只能等着，无法对解释器的资源进行访问，需要等待线程分配时间，这个线程把锁释放，另外的线程才开始运行。

三、异步

异步是相对于同步而言的，同步既指程序按照顺序一步一步往下执行，异步就是无序，无序等待上一步完成之后才可以执行下一步。

异步编程是一种并发编程的模式，其关注点是通过调度不同任务之间的执行和等待时间，通过减少处理器的闲置时间来达到减少整个程序的执行时间；异步编程跟同步编程模型最大的不同就是其任务的切换，当遇到一个需要等待长时间执行的任务的时候，我们可以切换到其他的任务执行。

asyncio 异步模块

async与await关键字：

  • async：定义异步
  • await：执行异步

相关函数：

  • gather：将异步函数批量执行，返回一个列表，既函数执行结果的列表
  • run：执行主异步函数，返回值是函数执行的返回值
import time, random


def zulu():
    for i in range(10):
        print('Zulu {}'.format(i))
        time.sleep(random.random() * 2)
    return 'zulu'


def tango():
    for i in range(10):
        print('Tango {}'.format(i))
        time.sleep(random.random() * 2)
    return 'tango'


if __name__ == '__main__':
    start = time.time()
    zulu()
    tango()
    end = time.time()
    print('耗时:{}'.format(end-start))

将同步改为异步执行的方式

async def zulu():
    for i in range(10):
        print('Zulu {}'.format(i))
        await asyncio.sleep(random.random() * 2)
    return 'zulu'


async def tango():
    for i in range(10):
        print('Tango {}'.format(i))
        await asyncio.sleep(random.random() * 2)
    return 'tango'


async def main():
    result = await asyncio.gather(
        zulu(),
        tango()
    )
    print(result)


if __name__ == '__main__':
    start = time.time()
    # zulu()
    # tango()
    asyncio.run(main())
    end = time.time()
    print('耗时:{}'.format(end-start))

zulu()和tango()两个函数交互执行，时间缩短一半

分别在zulu函数中和mian函数中打印出pid。

与多线程和多进程编程模型相比，异步编程只是在同一个线程之内的的任务调度

gevent 异步模块

gevent异步包需要通过pip进行安装

python3 -m pip install gevent -i https://pypi.tuna.tsinghua.edu.cn/simple

gevent 异步模块常用方法：

  • spawn：创建协程对象，参数为func以及传入函数的参数，返回一个协程对象
  • joinall：批量处理协程对象
  • get：获取函数返回结果
  • value：属性，也可以获取函数返回值
  • join：阻塞等待异步程序结束
  • kill：杀掉当前协程
  • dead：判断当前协程是否销毁
import time, random, os
import gevent


def zulu_gevent():
    for i in range(10):
        print('Zulu Gevent, PID：{}'.format(os.getpid()))
        gevent.sleep(random.random() ** 2)
    return 'Zulu with Gevent'


def tango_gevent():
    for i in range(10):
        print('Tango Gevent, PID：{}'.format(os.getpid()))
        gevent.sleep(random.random() ** 2)
    return 'Tango with Gevent'


if __name__ == '__main__':
    start = time.time()
    zulu = gevent.spawn(zulu_gevent)
    tango = gevent.spawn(tango_gevent)
    res_gevent = [zulu, tango]
    res = gevent.joinall(res_gevent)
    print(res)
    print(res[0].value)
    print(res[1].get())
    end = time.time()
    print('耗时：{}'.format(end - start))
    print('PID：{}'.format(os.getpid()))

调用value属性可以从协程对象中获取函数的返回值。

相关文章