本文目录导读：
- 使用
threading模块的线程优先级 - 使用
concurrent.futures模块 - 使用
asyncio实现协程优先级 - 使用
heapq实现优先级队列 - 使用第三方库
schedule实现定时优先级 - 完整的任务调度系统示例
- 选择建议
在Python中区分任务执行优先级有多种实现方式,我来介绍几种常用的方法：
使用 threading 模块的线程优先级
Python的线程实际上没有直接的优先级控制,但可以通过自定义实现：
import threading
import time
from queue import PriorityQueue
class PriorityTask:
def __init__(self, priority, name, func):
self.priority = priority # 数值越小优先级越高
self.name = name
self.func = func
def __lt__(self, other):
return self.priority < other.priority
class PriorityWorker(threading.Thread):
def __init__(self, queue):
super().__init__()
self.queue = queue
def run(self):
while True:
task = self.queue.get()
if task is None: # 停止信号
break
print(f"执行任务: {task.name}, 优先级: {task.priority}")
task.func()
self.queue.task_done()
def task_high():
time.sleep(0.5)
print("高优先级任务完成")
def task_medium():
time.sleep(1)
print("中优先级任务完成")
def task_low():
time.sleep(1.5)
print("低优先级任务完成")
# 使用示例
queue = PriorityQueue()
worker = PriorityWorker(queue)
worker.start()
# 添加任务
queue.put(PriorityTask(3, "低优先级", task_low))
queue.put(PriorityTask(1, "高优先级", task_high))
queue.put(PriorityTask(2, "中优先级", task_medium))
queue.join() # 等待所有任务完成
queue.put(None) # 停止工作线程
worker.join()
使用 concurrent.futures 模块
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
def task_with_priority(priority, task_id):
time.sleep(1)
return f"任务{task_id} (优先级{priority}) 完成"
# 创建线程池
with ThreadPoolExecutor(max_workers=3) as executor:
futures = []
# 添加不同优先级的任务
tasks = [
(1, "高优先级"),
(3, "低优先级"),
(2, "中优先级"),
(1, "另一个高优先级"),
]
for priority, name in tasks:
future = executor.submit(task_with_priority, priority, name)
futures.append((priority, future))
# 按优先级顺序处理结果
futures.sort(key=lambda x: x[0]) # 按优先级排序
for priority, future in futures:
result = future.result()
print(f"[优先级{priority}] {result}")
使用 asyncio 实现协程优先级
import asyncio
import random
class PriorityScheduler:
def __init__(self):
self.tasks = [] # 优先级队列
def add_task(self, priority, coro):
self.tasks.append((priority, coro))
self.tasks.sort(key=lambda x: x[0]) # 按优先级排序
async def run(self):
while self.tasks:
priority, coro = self.tasks.pop(0)
print(f"执行协程，优先级: {priority}")
await coro()
async def task_high():
await asyncio.sleep(0.5)
print("高优先级协程完成")
async def task_low():
await asyncio.sleep(1)
print("低优先级协程完成")
async def main():
scheduler = PriorityScheduler()
# 添加任务
scheduler.add_task(2, task_low())
scheduler.add_task(1, task_high())
scheduler.add_task(3, task_low()) # 另一个低优先级
await scheduler.run()
# 运行
asyncio.run(main())
使用 heapq 实现优先级队列
import heapq
import time
from threading import Thread
class PriorityTaskQueue:
def __init__(self):
self.heap = []
self._lock = threading.Lock()
def add_task(self, priority, task_id, task_func):
with self._lock:
heapq.heappush(self.heap, (priority, task_id, task_func))
def get_highest_priority_task(self):
with self._lock:
if self.heap:
return heapq.heappop(self.heap)[2] # 返回任务函数
return None
def worker(queue):
while True:
task = queue.get_highest_priority_task()
if task is None:
break
task()
time.sleep(0.1)
# 使用示例
import threading
task_queue = PriorityTaskQueue()
# 添加任务
task_queue.add_task(3, "task1", lambda: print("低优先级任务1"))
task_queue.add_task(1, "task2", lambda: print("高优先级任务2"))
task_queue.add_task(2, "task3", lambda: print("中优先级任务3"))
# 创建工作线程
worker_thread = threading.Thread(target=worker, args=(task_queue,))
worker_thread.start()
worker_thread.join(timeout=5)
使用第三方库 schedule 实现定时优先级
import schedule
import time
def high_priority_task():
print("执行高优先级任务")
def low_priority_task():
print("执行低优先级任务")
# 设置不同的执行频率来表示优先级
schedule.every(1).seconds.do(high_priority_task) # 高优先级：每秒执行
schedule.every(3).seconds.do(low_priority_task) # 低优先级：每3秒执行
while True:
schedule.run_pending()
time.sleep(0.1)
完整的任务调度系统示例
import threading
import queue
import time
from enum import Enum
from dataclasses import dataclass
from typing import Any, Callable
class Priority(Enum):
HIGH = 1
MEDIUM = 2
LOW = 3
def __lt__(self, other):
return self.value < other.value
@dataclass
class Task:
priority: Priority
name: str
func: Callable
args: tuple = ()
kwargs: dict = None
def __lt__(self, other):
return self.priority < other.priority
class TaskScheduler:
def __init__(self, num_workers: int = 3):
self.queue = queue.PriorityQueue()
self.workers = []
self.running = True
# 创建工作线程
for i in range(num_workers):
worker = threading.Thread(target=self._worker_loop, args=(i,))
worker.daemon = True
self.workers.append(worker)
worker.start()
def add_task(self, task: Task):
"""添加任务到调度器"""
self.queue.put(task)
print(f"添加任务: {task.name}, 优先级: {task.priority.name}")
def _worker_loop(self, worker_id: int):
"""工作线程主循环"""
while self.running:
try:
task = self.queue.get(timeout=1)
print(f"[工人{worker_id}] 执行任务: {task.name}, 优先级: {task.priority.name}")
# 执行任务
kwargs = task.kwargs or {}
try:
task.func(*task.args, **kwargs)
except Exception as e:
print(f"[工人{worker_id}] 任务{task.name}执行失败: {e}")
finally:
self.queue.task_done()
except queue.Empty:
continue
def stop(self):
"""停止调度器"""
self.running = False
for worker in self.workers:
worker.join(timeout=2)
# 使用示例
def process_data(data_id: int):
"""模拟数据处理任务"""
time.sleep(0.5)
print(f"处理数据: {data_id}")
# 创建调度器
scheduler = TaskScheduler(num_workers=2)
# 添加不同优先级的任务
tasks = [
Task(Priority.LOW, "低优先级任务1", process_data, args=(1,)),
Task(Priority.HIGH, "高优先级任务2", process_data, args=(2,)),
Task(Priority.MEDIUM, "中优先级任务3", process_data, args=(3,)),
Task(Priority.HIGH, "高优先级任务4", process_data, args=(4,)),
Task(Priority.LOW, "低优先级任务5", process_data, args=(5,)),
]
for task in tasks:
scheduler.add_task(task)
# 等待所有任务完成
time.sleep(5)
scheduler.stop()
选择建议
- 简单任务：使用
heapq+ 优先级队列 - I/O密集型：使用
asyncio协程 - CPU密集型：使用
multiprocessing加权队列 - 需要复杂调度：使用
APScheduler第三方库 - 微服务架构：使用
Celery或RQ
优先级实现的核心是定义清晰的优先级规则,并选择合适的队列和调度机制。
