二探并发(Actor)

上回谈到利用消息队列在线程（进程）间传递消息以此来实现线程间的通信。

共享的消息队列

这是建立在并发只是提高系统吞吐率的基础上，在这种假设下每一个worker所做的工作都一样，因此可以共享一个消息队列，谁拿到下一条指令都无所谓。

如果我们不是同胞呢

假如我们需要实现一个各个worker各司其职异步工作的系统（例如我们的IP-phone）该怎么办呢？假设我们还使用上面的方法，所有的worker共享一个消息队列，这时候就会产生问题：消息没办法发送给特定的worker。

怎么解决这个问题呢？一个简单粗暴的解决方案是给每个消息加上标识，以标识出这条消息所要发送的对象：

queue = Queue([('Send_to_worker_A', 'do_something1'),
               ('Send_to_worker_B', 'do_something2'),
               ('Send_to_worker_B', 'do_something3'),
               ('Send_to_worker_C', 'do_something4'),
               ('Send_to_worker_A', 'do_something5')])

看起来问题迎刃而解了对吧？但我们引入了一个新的问题。

如果我不小心拆开了送给别人的信

如果送给worker A的消息被worker B接收到了，他得到条消息其实并没有什么用，而本该得到这条消息的worker A却没得到它，从而这条信息就丢失了。既浪费了worker B的时间，由消耗了work A的青春，甚至有可能因为跳步而使得整个系统陷入到什么诡异的Bug中。

这个问题又怎么解决呢？从直觉上看，给每一个worker添加如下策略就能解决问题：得到别人的消息就把这条消息塞回消息队列中。

class Worker_B:
    
    def run(self):
        while True:
            if not self.queue.empty():
                data = self.queue.get()
                if data['to_whom'] == 'worker_B':
                    do_something_with_data
                else:
                    self.queue.put(data)

看似修修补补解决了问题…实际上又引入了一个新的问题。

这看起来是一条死路

第一，消息队列是一个队列，这意味着FIFO，就算引入了具有优先级的消息队列，本质上这一点也并没有改变。这样一来，本该第一个交给worker A的消息经worker B一倒腾，成为了当前消息队列中最晚递交给worker A的了，可能使得系统陷入诡异的Bug中。

第二，即使我们设法让分发错了的消息重新插回消息队列的队首，如果不引入锁机制，使得错收了消息worker B的收信-检查-放回操作成为一个原子操作的话，那么在work B执行放回操作的时候，可能其他worker已经在继续进行接收消息操作了，仍然有可能造成时序上的大问题，而引入锁机制索然能解决逻辑上的错误问题，但是这样会使得每次轮询中只有一个worker真正能够工作，而其他得到错误消息执行收信-检查-放回操作的worker白白浪费自己的时间片，效率低下。

所以，看起来针对为了异步工作而引入并发的系统来说，每个worker共享同一个消息队列并不是一个好办法。

根治这个问题的办法就是针对每一组执行同一任务的worker都建立一个专门的消息队列（邮箱），甚至每个worker都设置自己的消息队列，完全放弃了线程间共享内存的能力。后一种方法就是我今天要讲的主角：Actor模型（参与者模式）。

Actor模型

首先我们来看看维基百科对Actor模型的定义与概念：

概念

在计算机科学中，参与者模式（英语：Actor model）是一种并发运算上的模型。“参与者”是一种程序上的抽象概念，被视为并发运算的基本单元：当一个参与者接收到一则消息，它可以做出一些决策、创建更多的参与者、发送更多的消息、决定要如何回答接下来的消息。参与者模式在1973年于Carl Hewitt、Peter Bishop及Richard Steiger的论文中提出。

参与者模型推崇的哲学是“一切皆是参与者”，这与面向对象编程的“一切皆是对象”类似，但是面向对象编程通常是顺序执行的，而参与者模型是并行执行的。参与者是一个运算实体，回应接受到的消息，同时并行的：

• 发送有限数量的消息给其他参与者；
• 创建有限数量的新参与者；
• 指定接受到下一个消息时的行为。

以上操作不含有顺序执行的假设，因此可以并行进行。发送者与已经发送的消息解耦，是参与者模型的根本优势。这允许进行异步通信，同时满足消息传递的控制结构。消息接收者是通过地址区分的，有时也被称作“邮件地址”。因此参与者只能和它拥有地址的参与者通信。它可以通过接受到的信息获取地址，或者获取它创建的参与者的地址。参与者模型的特征是，参与者内部或之间进行并行计算，参与者可以动态创建，参与者地址包含在消息中，交互只有通过直接的异步消息通信，不限制消息到达的顺序。

实现

Actor足够简单，为并发而生，并且具有足够好的封装性可以隔离变化（例如：不关心是多线程还是多进程）。

让我们来看看一个典型的Actor的python线程实现：

from queue import Queue
from threading import Thread, Event


class WorkerExit(Exception):
    # 用来结束任务的异常
    pass


class BaseWorker(object):
	# Actor
    def __init__(self):
        self._mailbox = Queue()
        super(BaseWorker, self).__init__()

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

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

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

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

        t.daemon = True
        t.start()

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

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

    def run(self):
        raise NotImplementedError

简单地解释一下，这个Actor（我起名为BaseWorker，在真实的项目中作为父类使用），维持两个数据结构：Queue作为邮箱以及Event作为阻塞主线程的杀器。

这里作为外部接口的核心操作只有一个send()方法。请注意，我们不限制能够传递的消息类型，这意味着无比的灵活性。

在Actor内部我们使用一个线程来运行run()方法结合recv()方法执行所指定的具体工作。值得一提的是，我们设置了一个哨兵信号WorkerExit用来停止任务，请注意WorkerExit的运行原理是当他被识别后作为一个异常被抛出，而在异常处理这里我们甚至做更多事情，当然这里我们只是在捕获到这个异常时停之线程运行。而这一异常处理的实现有赖于用来包装run()方法的_bootstrap()。

在Actor模型的哲学下我们将这个简单的例子拓展从而走得更远。

总结

并发编程需要一种较之平常更为抽象的思维方式，也可以作为“高内聚，低耦合”这一思想的良好实践。

而我在这片对我而言的全新领域的探索中Actor的确是一盏指路明灯，算是领着我真正入了门，因此在这里分享。 :)

/python/ /并发/