本文目录导读：
- 使用
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
优先级实现的核心是定义清晰的优先级规则,并选择合适的队列和调度机制。
版权声明：NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
