Python线程与协程基础实践：从Thread到Queue及简单调度器

星野

发布时间：2026-10-13| 3人浏览过

原创

本文总结Python中线程与协程的基础用法。介绍利用threading库创建和管理线程，通过is_alive判断状态并调用_stop防止阻塞；探讨使用ThreadPool简化线程池操作。针对线程间通信，推荐使用Queue实现安全数据交换，并警告避免使用非put/get方法。此外，解析基于yield的单线程协程机制，通过TaskScheduler结合deque实现简单的任务调度，旨在提升程序性能与结构优化。

Python对线程的原生支持相对有限，但官方提供了完善的异步网络框架模块，未来值得深入探索。本文旨在对Python中的线程机制进行简要梳理与总结。

利用threading库可以在独立线程中执行任意Python可调用对象。尽管该模块对线程底层操作的支持有限，但仍可通过简单的线程处理I/O操作，从而有效降低程序的响应延迟。

from threading import Thread
import time
 
 
def countdown(n):
  while n > 0:
    print('T-minus:', n)
    n -= 1
 
 
t = Thread(target=countdown, args=(10,))
t.start() # 开启线程
 
time.sleep(2)
 
if t.is_alive() is True:
  print("停止线程...")
  t._stop() # 停止线程

start方法用于启动线程，而_stop方法用于终止线程。为避免线程在执行I/O操作时发生阻塞，建议在运行一段时间后检查线程存活状态；若线程仍在运行，则调用_stop()强制停止，以防止阻塞问题（可将_stop封装至类中，此处未作封装）。

相较于手动创建线程，使用ThreadPool线程池是更优的选择。若线程间无需共享变量，使用线程能显著减少复杂操作并节省时间。若需在线程间进行通信，可通过队列机制实现：

from queue import Queue
from threading import Thread
 
 
class kill:
  def terminate(self, t):
    if t.isAlive is True:
      t._stop()
 
 
def product(out_q):
  for i in range(5):
      out_q.put(i)
 
 
def consumer(in_q):
  for i in range(5):
    print(in_q.get())
 
 
q = Queue()
t1 = Thread(target=consumer, args=(q,))
t2 = Thread(target=product, args=(q,))
t1.start()
t2.start()
 
 
k = kill() # 查询线程是否终止，防止阻塞...
k.terminate(t1)
k.terminate(t2)

Queue实例由所有线程共享，并内置所需锁机制，因此可在任意数量的线程中安全共享。需注意，在多线程环境中，严禁使用put()和get()以外的Queue类方法，因其可靠性无法保证。对于小型线程间数据通信，队列是理想选择；若涉及大规模数据交互，Python提供了相关模块，具体实现需进一步查阅资料。

协程本质上是基于单线程环境的yield程序。

from collections import deque
 
 
def countdown(n):
  while n > 0:
    print("T-minus", n)
    yield # 返回之后下次直接从这里执行...相当于C#里面得yield return .
    n -= 1
  print("this is countdown!!!")
 
 
def countup(n):
  x = 0
  while x < n:
    print("Counting up", x)
    yield
    x += 1
 
 
class TaskScheduler:
  def __init__(self):
    self._task_queue = deque()
 
  def new_task(self, task):
    self._task_queue.append(task)
 
  def run(self):
    while self._task_queue:
      task = self._task_queue.popleft()
      try:
        next(task)
        self._task_queue.append(task)
      except StopIteration:
        pass
 
 
sche = TaskScheduler()
sche.new_task(countdown(10))
sche.new_task(countdown(5))
sche.new_task(countup(15))
sche.run()

结合近期使用Python的心得，Python虽性能常受诟病，但依然优秀。初学Python时，常尝试制作炫目的程序，如利用自然语言处理进行情感分析、编写热门爬虫程序或生成炫酷的数据分析图表。然而，随着深入，我意识到程序的核心不在于这些表面功能，只要掌握基本语法并阅读官方文档即可实现。代码的重点应在于性能与优化，即编写功能完善、性能最优且结构优美的程序。这正如老师所言的“文化软实力”，程序中的“软实力”体现在嵌入最合适的设计模式、执行完备的程序优化以及采用最节省性能的数据结构。

以上关于Python简单线程和协程的学习心得分享完毕，希望能对读者有所参考，也欢迎大家继续支持本站。