Python使用ThreadPoolExecutor线程池和Queue消息队列

admin2024-05-15  0

Python使用ThreadPoolExecutor线程池和Queue消息队列

1 介绍

(1)线程池

concurrent.futures模块是从 Python3.2 后引入的异步执行模块,主要用于在多核CPU和网络I/O中进行高效的并发编程。这个模块提供了ThreadPoolExecutor(线程池)和ProcessPoolExecutor(进程池)两个类。

可使用concurrent.futures.ThreadPoolExecutor直接创建线程池,需要设置最大的线程数,用submit()执行线程。

(2)消息队列

Python中的Queue是线程安全的消息队列,可使用queue.Queue完成消息队列。

put()和get()配合使用,可使用put()创建消息,使用get()消费消息,使用qsize()获取消息长度。

join()和task_done()配合使用,join()用于阻塞调用线程,直到队列中的所有任务被处理掉;task_done()用于通知队列任务已完成,每一个由put()调用入队的任务都有一个对应的task_done()调用。一般在使用join()阻塞线程时才会使用task_done()。

2 代码

import random
from concurrent.futures import ThreadPoolExecutor
from queue import Queue
from threading import Lock
from time import sleep

# 创建线程池
# max_workers:最大线程池的数量
max_workers = 3
executor = ThreadPoolExecutor(max_workers=max_workers)


# 添加锁
lock = Lock()

# 构建消息队列,Queue是线程安全的
queue_task = Queue()

# 数据库(假设)
db_record = list()


def run_task(msg):
    # 添加消息
    queue_task.put(msg)

    if queue_task.qsize() > max_workers:
        return

    # 执行线程池
    executor.submit(doing_task)


def doing_task():

    # 消费消息队列中的任务
    while queue_task.qsize() > 0:
        # 获取消息
        msg = queue_task.get()
        print("start-" + str(msg))

        # 跟新数据库(假设)
        # 添加锁
        lock.acquire()

        # 更新
        db_record.append(msg)

        # 释放锁
        lock.release()

        # 随机睡眠1-3秒
        sleep(random.randint(1, 3))

        print("end-" + str(msg))


if __name__ == '__main__':
    run_task(1)
    run_task(2)
    run_task(3)
    run_task(4)
    run_task(5)
    run_task(6)

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明原文出处。如若内容造成侵权/违法违规/事实不符,请联系SD编程学习网:675289112@qq.com进行投诉反馈,一经查实,立即删除!