线程创建

Python的多线程还是比较简单的，只需要定义一个工作函数，然后将工作函数传递给一个Thread 对象实例即可新建一个工作线程

from threading import Thread
import time

running_flag = True
def func(arg):
    while running_flag:
        print("Work in func... args: %d" % arg)
        time.sleep(1)

t = Thread(target=func, args=(1, ))
t.start()
time.sleep(10)
running_flag = False
t.join()

为了方便使用封装一个任务类来抽象多线程模型，我们仅需要编写loop函数来实现我们需要循环的功能函数。

from threading import Thread

class Task:
    def __init__(self) -> None:
        self._running = False
        self.t = None
    
    def terminate(self):
        self._running = False # set the _running flag to false
        if self.t is not None:
            self.t.join() # Join to wait the thread terminated

    def run(self):
        while self._running:
            self.loop()

    def loop(self): # thread loop function
        pass

    def start(self):
        self.t = Thread(target=self.run)
        self._running = True
        self.t.start()

使用示例如下：

from task_ import Task
import time
import sys

class IoTask(Task):
    def __init__(self) -> None:
        super().__init__()
    

    def loop(self):
        print('Running....')
        sys.stdout.flush()
        time.sleep(1.0)

task = IoTask()
task.start()
time.sleep(10)
task.terminate()

如果正常工作的话，我们可以看到十次Running.... 输出

Running....
Running....
Running....
Running....
Running....
Running....
Running....
Running....
Running....
Running....

线程间的数据交互

多个线程间的数据通信可以依赖一些线程安全的数据结构比如Python 系统库中的队列数据结构Queue

from task_ import Task
from queue import Queue
import time
import sys

class IoQueueTask(Task):
    def __init__(self) -> None:
        super().__init__()
        self.queue = Queue()

    def loop(self):
        if not self.queue.empty():
            msg = self.queue.get()
            print('Get message from queue: %s' %msg)
        else:
            time.sleep(1.0)

task = IoQueueTask()
task.start()
for i in range(10):
    task.queue.put('Message - %d' % i)
    if i > 5:
        time.sleep(10)
task.terminate()

参考

  1. Python-Cookbook启动与停止线程
  2. Python-Cookbook线程间通信