并发编程

启动与停止线程

问题：

　　你要为需要并发执行的代码创建/销毁线程 

解决方案：

　　threading 库可以在单独的线程中执行任何的在Python 中可以调用的对象。你可以创建一个Thread 对象并将你要执行的对象以target 参数的形式提供给该对象。下面是一个简单的例子：

import time
from threading import Thread

def countdown(n):
    while n > 0:
        print('T-minus', n)
        n -= 1
        time.sleep(5)

t = Thread(target=countdown, args=(10,))
t.start()

以上代码执行的结果为：

T-minus 10
T-minus 9
T-minus 8
T-minus 7
T-minus 6
T-minus 5
T-minus 4
T-minus 3
T-minus 2
T-minus 1

　　当你创建好一个线程对象后，该对象并不会立即执行，除非你调用它的start()方法（当你调用start() 方法时，它会调用你传递进来的函数，并把你传递进来的参数传递给该函数）。Python 中的线程会在一个单独的系统级线程中执行（比如说一个POSIX 线程或者一个Windows 线程），这些线程将由操作系统来全权管理。线程一旦启动，将独立执行直到目标函数返回。你可以查询个线程对象的状态，看它是否还在执行：

if t.is_alive():
    print('Still running')
else:
    print('Completed')

你也可以将一个线程加入到当前线程，并等待它终止：

t.join()

Python 解释器在所有线程都终止后才继续执行代码剩余的部分。对于需要长时间运行的线程或者需要一直运行的后台任务，你应当考虑使用后台线程。例如：

 

t = Thread(target=countdown, args=(10,), daemon=True)
t.start()

　　后台线程无法等待，不过，这些线程会在主线程终止时自动销毁。除了如上所示的两个操作，并没有太多可以对线程做的事情。你无法结束一个线程，无法给它发送号，无法调整它的调度，也无法执行其他高级操作。如果需要这些特性，你需要自己添加。比如说，如果你需要终止线程，那么这个线程必须通过编程在某个特定点轮询来退出。你可以像下边这样把线程放入一个类中：

import time
from threading import Thread


class CountdownTask:
    def __init__(self):
        self._running = True

    def terminate(self):
        self._running = False

    def run(self, n):
        while self._running and n > 0:
            print('T-minus', n)
            n -= 1
            time.sleep(5)

if __name__ == '__main__':
    c = CountdownTask()
    t = Thread(target=c.run, args=(10,))
    t.start()
    c.terminate()
    t.join()

　　如果线程执行一些像I/O 这样的阻塞操作，那么通过轮询来终止线程将使得线程之间的协调变得非常棘手。比如，如果一个线程一直阻塞在一个I/O 操作上，它就永远无法返回，也就无法检查自己是否已经被结束了。要正确处理这些问题，你需要利用超时循环来小心操作线程。例子如下：

class IOTask:
    def terminate(self):
        self._running = False

    def run(self, sock):
        #设置超时时间
        sock.settimeout(5)
        while self._running:
            try:
                data = sock.recv(8192)
                break
            except Exception as e:
                continue

            pass
        return 

注意：

　　由于全局解释锁（GIL）的原因，Python 的线程被限制到同一时刻只允许一个线程执行这样一个执行模型。所以，Python 的线程更适用于处理I/O 和其他需要并发执行的阻塞操作（比如等待I/O、等待从数据库获取数据等等），而不是需要多处理器并行的计算密集型任务。

　　有时你会看到下边这种通过继承Thread 类来实现的线程：

import time
from threading import Thread


class CountdownThread(Thread):

    def __init__(self, n):
        super().__init__()
        self.n = n

    def run(self):
        while self.n > 0:
            print('T-minus', self.n)
            self.n -= 1
            time.sleep(5)

c = CountdownThread(5)
c.start()

　　尽管这样也可以工作，但这使得你的代码依赖于threading 库，所以你的这些代码只能在线程上下文中使用。上文所写的那些代码、函数都是与threading 库无关的，这样就使得这些代码可以被用在其他的上下文中，可能与线程有关，也可能与线程无关。比如，你可以通过multiprocessing 模块在一个单独的进程中执行你的代码：

import time
import multiprocessing
from threading import Thread


class CountdownThread(Thread):

    def __init__(self, n):
        super().__init__()
        self.n = n

    def run(self):
        while self.n > 0:
            print('T-minus', self.n)
            self.n -= 1
            time.sleep(5)

c = CountdownThread(5)
p = multiprocessing.Process(target=c.run)
p.start()

再次重申，这段代码仅适用于CountdownTask 类是以独立于实际的并发手段（多线程、多进程等等）实现的情况。

 

判断线程是否已经启动

问题：

　　你已经启动了一个线程，但是你想知道它是不是真的已经开始运行了

解决方案：

　　线程的一个关键特性是每个线程都是独立运行且状态不可预测。如果程序中的其他线程需要通过判断某个线程的状态来确定自己下一步的操作，这时线程同步问题就会变得非常棘手。为了解决这些问题，我们需要使用threading 库中的Event 对象。Event 对象包含一个可由线程设置的信号标志，它允许线程等待某些事件的发生。在初始情况下，event 对象中的信号标志被设置为假。如果有线程等待一个event 对象，而这个event 对象的标志为假，那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个event 对象的信号标志设置为真，它将唤醒所有等待这个event 对象的线程。如果一个线程等待一个已经被设置为真的event 对象，那么它将忽略这个事件，继续执行。下边的代码展示了如何使用Event 来协调线程的启动：

from threading import Thread, Event
import time


def countdown(n, started_evt):
    print('countdown starting')
    started_evt.set()
    while n > 0:
        print('T-minus', n)
        n -= 1
        time.sleep(5)

started_evt = Event()

print('Launching countdown')
t = Thread(target=countdown, args=(10, started_evt))
t.start()
started_evt.wait()
print('countdown is running')

以上代码执行的结果为：

Launching countdown
countdown starting
T-minus 10
countdown is running
T-minus 9
T-minus 8
T-minus 7
T-minus 6
T-minus 5
T-minus 4
T-minus 3
T-minus 2
T-minus 1

　　当你执行这段代码，“countdown is running”总是显示在“countdown starting”之后显示。这是由于使用event 来协调线程，使得主线程要等到countdown() 函数输出启动信息后，才能继续执行。

 

 

线程间通信

问题：

　　你的程序中有多个线程，你需要在这些线程之间安全地交换信息或数据

解决方案：

　　从一个线程向另一个线程发送数据最安全的方式可能就是使用queue 库中的队列了。创建一个被多个线程共享的Queue 对象，这些线程通过使用put() 和get() 操作来向队列中添加或者删除元素。例如：

　　Queue 对象已经包含了必要的锁，所以你可以通过它在多个线程间多安全地共享数据。当使用队列时，协调生产者和消费者的关闭问题可能会有一些麻烦。一个通用的解决方法是在队列中放置一个特殊的值，当消费者读到这个值的时候，终止执行。例如：

import heapq
import threading


class PriorityQueue:
    def __init__(self):
        self._queue = []
        self._count = 0
        self._cv = threading.Condition()

    def put(self, item, priority):
        heapq.heappush(self._queue, (-priority, self._count, item))
        self._count += 1
        self._cv.notify()

    def get(self):
        with self._cv:
            while len(self._queue) == 0:
                self._cv.wait()
            return heapq.heappop(self._queue)[-1]
    

　　使用队列来进行线程间通信是一个单向、不确定的过程。通常情况下，你没有办法知道接收数据的线程是什么时候接收到的数据并开始工作的。不过队列对象提供一些基本完成的特性，比如下边这个例子中的task done() 和join() ：

from queue import Queue
from threading import Thread


# A thread that produces data
def producer(out_q):
    while running:
        # Produce some data
        ...
        out_q.put(data)


# A thread that consumes data
def consumer(in_q):
    while True:
        # Get some data
        data = in_q.get()
        # Process the data
        ...
        # Indicate completion
        in_q.task_done()

# Create the shared queue and launch both threads
q = Queue()
t1 = Thread(target=consumer, args=(q,))
t2 = Thread(target=producer, args=(q,))
t1.start()
t2.start()
# Wait for all produced items to be consumed
q.join()

　　如果一个线程需要在一个“消费者”线程处理完特定的数据项时立即得到通知，你可以把要发送的数据和一个Event 放到一起使用，这样“生产者”就可以通过这个Event 对象来监测处理的过程了。示例如下：

from queue import Queue
from threading import Thread, Event

# A thread that produces data
def producer(out_q):
    while running:
        # Produce some data
        ...
        # Make an (data, event) pair and hand it to the consumer
        evt = Event()
        out_q.put((data, evt))
        ...
        # Wait for the consumer to process the item
        evt.wait()

# A thread that consumes data
def consumer(in_q):
    while True:
        # Get some data
        data, evt = in_q.get()
        # Process the data
        ...
        # Indicate completion
        evt.set()
        evt.set()

给关键部分加锁

问题：

　　你需要对多线程程序中的临界区加锁以避免竞争条件

解决方案：

　　要在多线程程序中安全使用可变对象，你需要使用threading 库中的Lock 对象，就像下边这个例子这样：

import threading


class SharedCounter:

    def __init__(self, initial_value=0):
        self._value = initial_value
        self._value_lock = threading.Lock()

    def incr(self, deltr=1):
        with self._value_lock:
            self._value += deltr

    def decr(self, delta=1):
        with self._value_lock:
            self._value -= delta

　　Lock 对象和with 语句块一起使用可以保证互斥执行，就是每次只有一个线程可以执行with 语句包含的代码块。with 语句会在这个代码块执行前自动获取锁，在执行结束后自动释放锁。

　　线程调度本质上是不确定的，因此，在多线程程序中错误地使用锁机制可能会导致随机数据损坏或者其他的异常行为，我们称之为竞争条件。为了避免竞争条件，最好只在临界区（对临界资源进行操作的那部分代码）使用锁。在一些“老的”Python 代码中，显式获取和释放锁是很常见的。下边是一个上一个例子的变种：

import threading

class SharedCounter:

    def __init__(self, initial_value=0):
        self._value = initial_value
        self._value_lock = threading.Lock()

    def incr(self, delta=1):
        self._value_lock.acquire()
        self._value += delta
        self._value_lock.release()

    def decr(self, delta=1):
        self._value_lock.acquire()
        self._value -= delta
        self._value_lock.release()

　　相比于这种显式调用的方法，with 语句更加优雅，也更不容易出错，特别是程序员可能会忘记调用release() 方法或者程序在获得锁之后产生异常这两种情况（使用with语句可以保证在这两种情况下仍能正确释放锁）。为了避免出现死锁的情况，使用锁机制的程序应该设定为每个线程一次只允许获取一个锁。如果不能这样做的话，你就需要更高级的死锁避免机制。在threading 库中还提供了其他的同步原语，比如RLoct 和Semaphore 对象。但是根据以往经验，这些原语是用于一些特殊的情况，如果你只是需要简单地对可变对象进行锁定，那就不应该使用它们。一个RLock （可重入锁）可以被同一个线程多次获取，主要用来实现基于监测对象模式的锁定和同步。在使用这种锁的情况下，当锁被持有时，只有一个线程可以使用完整的函数或者类中的方法。比如，你可以实现一个这样的SharedCounter 类：

import threading


class SharedCounter:

    _lock = threading.RLock()
    def __init__(self, initial_value=0):
        self._value = initial_value

    def incr(self, delta=1):
        with SharedCounter._lock:
            self._value += delta

    def decr(self, delta=1):
        with SharedCounter._lock:
            self.incr(-delta)

　　在上边这个例子中，没有对每一个实例中的可变对象加锁，取而代之的是一个被所有实例共享的类级锁。这个锁用来同步类方法，具体来说就是，这个锁可以保证一次只有一个线程可以调用这个类方法。不过，与一个标准的锁不同的是，已经持有这个锁的方法在调用同样使用这个锁的方法时，无需再次获取锁。比如decr 方法。这种实现方式的一个特点是，无论这个类有多少个实例都只用一个锁。因此在需要大量使用计数器的情况下内存效率更高。不过这样做也有缺点，就是在程序中使用大量线程并频繁更新计数器时会有争用锁的问题。信号量对象是一个建立在共享计数器基础上的同步原语。如果计数器不为0，with 语句将计数器减1，线程被允许执行。with 语句执行结束后，计数器加１。如果计数器为0，线程将被阻塞，直到其他线程结束将计数器加1。尽管你可以在程序中像标准锁一样使用信号量来做线程同步，但是这种方式并不被推荐，因为使用信号量为程序增加的复杂性会影响程序性能。相对于简单地作为锁使用，信号量更适用于那些需要在线程之间引入信号或者限制的程序。比如，你需要限制一段代码的并发访问量，你就可以像下面这样使用信号量完成：

from threading import Semaphore
from urllib.request import urlopen

_fetch_url_sema = Semaphore(5)

def fetch_url_sema(url):
    with _fetch_url_sema:
        return urlopen(url)

保存线程的状态信息

问题：

　　你需要保存正在运行线程的状态，这个状态对于其他的线程是不可见的

解决方案：

　　有时在多线程编程中，你需要只保存当前运行线程的状态。要这么做，可使用thread.local() 创建一个本地线程存储对象。对这个对象的属性的保存和读取操作都只会对执行线程可见，而其他线程并不可见。

from socket import socket, AF_INET, SOCK_STREAM
import threading


class LazyConnection:
    
    def __init__(self, address, family=AF_INET, type=SOCK_STREAM):
        self.address = address
        self.family = family
        self.type = type
        self.local = threading.local()

    def __enter__(self):
        if hasattr(self.local, 'sock'):
            raise  RuntimeError('Already connected')
        self.local.sock = socket(self.family, self.type)
        self.local.sock.connect(self.address)
        return self.local.sock

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.local.sock.close()
        del self.local.sock

　　代码中， 自己观察对于self.local 属性的使用。它被初始化尾一个threading.local() 实例。其他方法操作被存储为self.local.sock 的套接字对象。有了这些就可以在多线程中安全的使用LazyConnection 实例了。例如： 

from functools import partial


def test(conn):
    with conn as s:
        s.send(b'GET /index.html HTTP/1.0
')
        s.send(b'Host: www.python.org
')
        s.send(b'
')
        resp = b''.join(iter(partial(s.recv, 8192), b''))
    print('Got {} bytes'.format(len(resp)))

if __name__ == '__main__':
    conn = LazyConnection(('www.python.org', 80))
    t1 = threading.Thread(target=test, args=(conn,))
    t2 = threading.Thread(target=test, args=(conn,))
    t1.start()
    t2.start()
    t1.join()
    t2.join()

创建一个线程池

问题：

　　你创建一个工作者线程池，用来相应客户端请求或执行其他的工作

解决方案：

　　 concurrent.futures 函数库有一个ThreadPoolExecutor 类可以被用来完成这个任务。下面是一个简单的TCP 服务器，使用了一个线程池来响应客户端：

from socket import socket, AF_INET, SOCK_STREAM
from concurrent.futures import ThreadPoolExecutor


def echo_client(sock, client_addr):
    print('Got connection from', client_addr)
    while True:
        msg = sock.recv(65536)
        if not msg:
            break
        sock.sendall(msg)
    print('Client closed connection')
    sock.close()

def echo_server(addr):
    pool = ThreadPoolExecutor(128)
    sock = socket(AF_INET, SOCK_STREAM)
    sock.bind(addr)
    sock.listen(5)
    while True:
        client_sock, client_addr = sock.accept()
        pool.submit(echo_client, client_sock, client_addr)


echo_server(('',15000))

如果你想手动创建你自己的线程池，通常可以使用一个Queue 来轻松实现。下面是一个稍微不同但是手动实现的例子：

from socket import socket, AF_INET, SOCK_STREAM
from threading import Thread
from queue import Queue


def echo_client(q):
    sock, client_addr = q.get()
    print('Got connection from', client_addr)
    while True:
        msg = sock.recv(65536)
        if not msg:
            break
    sock.sendall(msg)
    print('Client closed connection')

    sock.close()


def echo_server(addr, nworkers):
    q = Queue()
    for n in range(nworkers):
        t = Thread(target=echo_client, args=(q, ))
        t.daemon = True
        t.start()

    sock = socket(AF_INET, SOCK_STREAM)
    sock.bind(addr)
    sock.listen(5)
    while True:
        client_sock, client_addr = sock.accept()
        q.put((client_sock, client_addr))


echo_server(('', 15000), 128)

使用ThreadPoolExecutor 相对于手动实现的一个好处在于它使得任务提交者更方便的从被调用函数中获取返回值。例如，你可能会像下面这样写：

from concurrent.futures import ThreadPoolExecutor
from urllib.request import urlopen
from ssl import PROTOCOL_SSLv23, SSLContext

def fetch_url(url):
    gcontext = SSLContext(PROTOCOL_SSLv23)
    u = urlopen(url, context=gcontext)
    data = u.read().decode('utf-8')
    return data

pool = ThreadPoolExecutor(10)

a = pool.submit(fetch_url, 'https://www.python.org')
b = pool.submit(fetch_url, 'http://www.pypy.org')

x = a.result()
y = b.result()

例子中返回的handle 对象会帮你处理所有的阻塞与协作，然后从工作线程中返回数据给你。特别的，a.result() 操作会阻塞进程直到对应的函数执行完成并返回一个结果。

简单的并行编程

问题：

　　你有个程序要执行CPU 密集型工作，你想让他利用多核CPU 的优势来运行的快一点

解决方案： 

　　concurrent.futures 库提供了一个ProcessPoolExecutor 类，可被用来在一个单独的Python 解释器中执行计算密集型函数。不过，要使用它，你首先要有一些计算密集型的任务。我们通过一个简单而实际的例子来演示它。假定你有个Apache web 服务器日志目录的gzip 压缩包：

logs/
    20170701.log.gz
    20170702.log.gz
    20170703.log.gz
    20170704.log.gz
    20170705.log.gz
    20170706.log.gz
    ...

进一步假设每个日志文件内容类似下面这样：

124.115.6.12 - - [10/Jul/2017:00:18:50 -0500] "GET /robots.txt ..." 200 71
210.212.209.67 - - [10/Jul/2017:00:18:51 -0500] "GET /ply/ ..." 200 11875
210.212.209.67 - - [10/Jul/2017:00:18:51 -0500] "GET /favicon.ico ..." 404 369
61.135.216.105 - - [10/Jul/2017:00:20:04 -0500] "GET /blog/atom.xml ..." 304 -

下面是一个脚本，在这些日志文件中查找出所有访问过robots.txt 文件的主机：

# findrobots.py

import gzip
import io
import glob

def find_robots(filename):

    robots = set()
    with gzip.open(filename) as f:
        for line in io.TextIOWrapper(f, encoding='utf-8'):
            fields = line.split()
            if fields[6] == '/robots.txt':
                robots.add(fields[0])

    return robots


def find_all_robots(logdir):
    files = glob.glob(logdir + '/*.log.gz')
    all_robots = set()
    for robots in map(find_robots, files):
        all_robots.update(robots)
    return all_robots

if __name__ == '__main__':
    robots = find_all_robots('logs')
    for ipaddr in robots:
        print(ipaddr)

　　前面的程序使用了通常的map-reduce 风格来编写。函数find robots() 在一个文件名集合上做map 操作，并将结果汇总为一个单独的结果，也就是find all robots()函数中的all robots 集合。现在，假设你想要修改这个程序让它使用多核CPU。很简单——只需要将map() 操作替换为一个concurrent.futures 库中生成的类似操作即可。下面是一个简单修改版本：

import gzip
import io
import glob
from concurrent import futures


def find_robots(filename):
    robots = set()
    with gzip.open(filename) as f:
        for line in io.TextIOWrapper(f, encoding='ascii'):
            fields = line.split()
            if fields[6] == '/robots.txt':
                robots.add(fields[0])

    return robots


def find_all_robots(logdir):
    files = glob.glob(logdir+'/*.log.gz')
    all_robots = set()

    with futures.ProcessPoolExecutor() as pool:
        for robots in pool.map(find_robots, files):
            all_robots.update(robots)

    return all_robots

if __name__ == "__main__":
    robots = find_all_robots('logs')
    for ipaddr in robots:
        print(ipaddr)

通过这个修改后，运行这个脚本产生同样的结果，但是在四核机器上面比之前快了3.5 倍。实际的性能优化效果根据你的机器CPU 数量的不同而不同。

 

 

定义一个Actor 任务

问题：

　　你想定义跟actor 模式中类似“actors”角色的任务

解决方案：

　　actore 模式是一种最古老的也是最简单的并行和分布式计算解决方案。事实上，它天生的简单性是它如此受欢迎的重要原因之一。简单来讲，一个actor 就是一个并发执行的任务，只是简单的执行发送给它的消息任务。响应这些消息时，它可能还会给其他actor 发送更进一步的消息。actor 之间的通信是单向和异步的。因此，消息发送者不知道消息是什么时候被发送，也不会接收到一个消息已被处理的回应或通知。

　　结合使用一个线程和一个队列可以很容易的定义actor，例如：

 

from queue import Queue
from threading import Thread, Event

class ActorExit(Exception):
    pass

class Actor:

    def __init__(self):
        self._mailbox = Queue()

    def send(self, msg):
        self._mailbox.put(msg)

    def recv(self):
        msg = self._mailbox.get()
        if msg is ActorExit:
            raise ActorExit
        return msg

    def close(self):
        self.send(ActorExit)

    def start(self):
        self._terminated = Event()
        t = Thread(target=self._bootstrap)
        t.daemon = True
        t.start()

    def _bootstrap(self):
        try:
            self.run()
        except ActorExit:
            pass
        finally:
            self._terminated.set()

    def join(self):
        self._terminated.wait()

    def run(self):
        while True:
            msg = self.recv()


class PrintActor(Actor):
    def run(self):
        while True:
            msg = self.recv()
            print('Got:', msg)

p = PrintActor()
p.start()
p.send('Hello')
p.send('World')
p.close()
p.join()

　　这个例子中，你使用actor 实例的send() 方法发送消息给它们。其机制是，这个方法会将消息放入一个队里中，然后将其转交给处理被接受消息的一个内部线程。close() 方法通过在队列中放入一个特殊的哨兵值（ActorExit）来关闭这个actor。用户可以通过继承Actor 并定义实现自己处理逻辑run() 方法来定义新的actor。ActorExit 异常的使用就是用户自定义代码可以在需要的时候来捕获终止请求（异常被get() 方法抛出并传播出去。

　　如果你放宽对于同步和异步消息发送的要求，类actor 对象还可以通过生成器来简化定义。例如：

def print_actor():
    while True:
        try:
            msg = yield # Get a message
            print('Got:', msg)
        except GeneratorExit:
            print('Actor terminating')

# Sample use
p = print_actor()
next(p) # Advance to the yield (ready to receive)
p.send('Hello')
p.send('World')
p.close()

实现消息发布/订阅模型

问题：

　　你有一个基于线程通信的程序，想让它们实现发布/订阅模式的消息通信

解决方案： 

　　要实现发布/订阅的消息通信模式，你通常要引入一个单独的“交换机”或“网关”对象作为所有消息的中介。也就是说，不直接将消息从一个任务发送到另一个，而是将其发送给交换机，然后由交换机将它发送给一个或多个被关联任务。下面是一个非常简单的交换机实现例子：

from collections import defaultdict


class Exchange:

    def __init__(self):
        self._subscribers = set()

    def attch(self, task):
        self._subscribers.add(task)

    def detach(self, task):
        self._subscribers.remove(task)

    def send(self, msg):
        for subscriber in self._subscribers:
            subscriber.send(msg)

_exchanges = defaultdict(Exchange)


def get_exchange(name):
    return _exchanges[name]

下面是一个简单例子，演示了如何使用一个交换机：

# Example of a task. Any object with a send() method
class Task:
    ...
    def send(self, msg):
        ...
task_a = Task()
task_b = Task()

# Example of getting an exchange
exc = get_exchange('name')

# Examples of subscribing tasks to it
exc.attach(task_a)
exc.attach(task_b)

# Example of sending messages
exc.send('msg1')
exc.send('msg2')

# Example of unsubscribing
exc.detach(task_a)
exc.detach(task_b)

使用生成器代替线程

问题：

　　你想使用生成器（协程）替代系统线程来实现并发。这个有时又被称为用户级线程或绿色线程

解决方案：

　　要使用生成器实现自己的并发，你首先要对生成器函数和yield 语句有深刻理解。yield 语句会让一个生成器挂起它的执行，这样就可以编写一个调度器，将生成器当做某种“任务”并使用任务协作切换来替换它们的执行。要演示这种思想，考虑下面两个使用简单的yield 语句的生成器函数：

def countdown(n):
    while n > 0:
        print('T-minus', n)
        yield
        n -= 1
    print('Blastoff!')


def countup(n):
    x = 0
    while x < n:
        print('Counting up', x)
        yield
        x += 1

这些函数在内部使用yield 语句，下面是一个实现了简单任务调度器的代码：

from collections import deque


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

sched = TaskScheduler()
sched.new_task(countdown(10))
sched.new_task(countdown(5))
sched.new_task(countup(15))
sched.run()

TaskScheduler 类在一个循环中运行生成器集合——每个都运行到碰到yield 语句为止。运行这个例子，输出如下：

T-minus 10
T-minus 5
Counting up 0
T-minus 9
T-minus 4
Counting up 1
T-minus 8
T-minus 3
Counting up 2
T-minus 7
T-minus 2
Counting up 3
T-minus 6
T-minus 1
Counting up 4
T-minus 5
Blastoff!
Counting up 5
T-minus 4
Counting up 6
T-minus 3
Counting up 7
T-minus 2
Counting up 8
T-minus 1
Counting up 9
Blastoff!
Counting up 10
Counting up 11
Counting up 12
Counting up 13
Counting up 14

下面的代码演示了使用生成器来实现一个不依赖线程的actor：

from collections import deque


class ActorScheduler:
    def __init__(self):
        self._actors = {}
        self._msg_queue = deque()

    def new_actor(self, name, actor):
        self._msg_queue.append((actor, None))
        self._actors[name] = actor

    def send(self, name, msg):
        actor = self._actors.get(name)
        if actor:
            self._msg_queue.append((actor, msg))

    def run(self):
        while self._msg_queue:
            actor, msg = self._msg_queue.popleft()
            try:
                actor.send(msg)
            except StopIteration:
                pass

if __name__ == "__main__":
    def printer():
        while True:
            msg = yield
            print('Got', msg)

    def counter(sched):
        while True:
            n = yield
            if n == 0:
                break
            sched.send('printer', n)
            sched.send('counter', n-1)


    sched = ActorScheduler()

    # Create the initial actors
    sched.new_actor('printer', printer())
    sched.new_actor('counter', counter(sched))

    # Send an initial message to the counter to initiate
    sched.send('counter', 10000)
    sched.run()

多个线程队列轮询

问题：

　　你有一个线程队列集合，想为到来的元素轮询它们，就跟你为一个客户端请求去轮询一个网络连接集合的方式一样

解决方案：

　　对于轮询问题的一个常见解决方案中有个很少有人知道的技巧，包含了一个隐藏的回路网络连接。本质上讲其思想就是：对于每个你想要轮询的队列，你创建一对连接的套接字。然后你在其中一个套接字上面编写代码来标识存在的数据，另外一个套接字被传给select() 或类似的一个轮询数据到达的函数。下面的例子演示了这个思想：

import queue
import socket
import os


class PollableQueue(queue.Queue):
    def __init__(self):
        super().__init__()
        if os.name == 'posix':
            self._putsocket, self._getsocket = socket.socketpair()
        else:
            server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            server.bind(('127.0.0.1', 0))
            server.listen(1)
            self._putsocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self._putsocket.connect(server.getsockname())
            self._getsocket, _ = server.accept()
            server.close()

    def fileno(self):
        return self._getsocket.fileno()

    def put(self, item):
        super().put(item)
        self._putsocket.send(b'x')

    def get(self):
        self._getsocket.recv(1)
        return super().get()
    

　　在这个代码中，一个新的Queue 实例类型被定义，底层是一个被连接套接字对。在Unix 机器上的socketpair() 函数能轻松的创建这样的套接字。在Windows 上面，你必须使用类似代码来模拟它。然后定义普通的get() 和put() 方法在这些套接字上面来执行I/O 操作。put() 方法再将数据放入队列后会写一个单字节到某个套接字中去。而get() 方法在从队列中移除一个元素时会从另外一个套接字中读取到这个单字节数据。

　　fileno() 方法使用一个函数比如select() 来让这个队列可以被轮询。它仅仅只是暴露了底层被get() 函数使用到的socket 的文件描述符而已。

　　下面是一个例子，定义了一个为到来的元素监控多个队列的消费者：

import select
import threading


def consumer(queues):
    while True:
        can_read, _, _ = select.select(queues, [], [])
        for r in can_read:
            item = r.get()
            print('Got:', item)


q1 = PollableQueue()
q2 = PollableQueue()
q3 = PollableQueue()
t = threading.Thread(target=consumer, args=([q1,q2,q3],))
t.daemon = True
t.start()

# Feed data to the queues
q1.put(1)
q2.put(10)
q3.put('hello')
q2.put(15)