python actor_python基础进阶：分布式计算解决方案actor，逐个击破
本文的文字及图片来源于网络,仅供学习、交流使用,不具有任何商业用途,版权归原作者所有,如有问题请及时联系我们以作处理
本文章来自腾讯云 作者：我是李超人哔哩哔哩 ( ゜- ゜)つロ 乾杯~ Bilibilispace.bilibili.com
actor模式是一种最古老的也是最简单的并行和分布式计算解决方案。一个actor就是一个并发执行的任务，只是简单的执行发送给它的消息任务。actor之间的通信是单向和异步的。因此，消息发送者不知道消息是什么时候被发送， 也不会接收到一个消息已被处理的回应或通知。
使用线程加队列可以定义一个actor：
from queue import Queue
from threading import Thread, Event
import time
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):
'''关闭actor'''
self.send(ActorExit)
def start(self):
'''启动actor'''
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):
'''消费者线程的run方法'''
while True:
msg = self.recv()
class PrintActor(Actor):
def run(self):
while True:
msg = self.recv()
print('Got:', msg)
p = PrintActor()
p.start()
time.sleep(1)
p.send('Hello')
time.sleep(2)
p.send('World')
time.sleep(1)
p.close()
p.join()
输出：
Got: Hello
Got: World
这个案例是这样的，先定义了一个actor，然后通过继承actor定义了PrintActor，并重写了其中的run方法。
这个actor中的每个方法起到什么作用，并如何实现的？
Start：创建self._terminated信号(关于event我前面的文章有讲解)，定义一个线程，将现场设置为守护线程，并启动线程。
_bootstrap：线程启动后会执行这个方法，并启动里面的run方法。
Run：这里的run方法被子类重写了，子类通过recv不断的获取队列中的消息。
Recv：获取队列中的消息，并返回消息，如果消息类型是ActorExit，则抛出ActorExit异常。
Send：往队列中写入消息
Close：发送一个ActorExit类型的消息，recv接受到这个消息后，抛出ActorExit异常，run方法结束，bootstrap中捕获这个异常后，将self._terminated信号修改为true。
Join：如果self._terminated为false线程阻塞。
思考：在程序最后面加了一个p.join()，为何要加这个呢？
因为close方法只是发送了一个ActorExit类型的消息到队列，而消费队列的线程是一个守护线程，如果close后面没有任何需要执行的代码，则主线程就结束了，子线程也会随着结束，那么很有可能‘world’这条消息也没消费到，子线程就退出了，所以加个p.join()是为了主线程被阻塞，知道子线程消费到ActorExit类型的消息，将self._terminated设置为true才会唤醒主线程，主线程退出，子线程也退出，此时队列中的消息肯定是已经消费完了。
上面这个案例中，actor只是简单的模拟了一个队列传递消息的例子，实际上这些消息还可以是一个函数，比如下面这个案例：
from threading import Event
class Result:
def __init__(self):
self._evt = Event()
self._result = None
def set_result(self, value):
self._result = value
self._evt.set()
def result(self):
self._evt.wait()
return self._result
class Worker(Actor):
def submit(self, func, *args, **kwargs):
r = Result()
self.send((func, args, kwargs, r))
return r
def run(self):
while True:
func, args, kwargs, r = self.recv()
r.set_result(func(*args, **kwargs))
def add(a, b):
return a+b
worker = Worker()
worker.start()
r = worker.submit(add, 2, 3)
worker.close()
worker.join()
print(r.result())
输出：
5
这个案例中，worker继承了actor，在submit方法中，将函数和函数的参数作为一个元祖发送到队列中，并返回一个Result实例，消费者线程拿到函数和参数后，执行这个函数，并将函数的返回值放到Result实例的_result属性上。_result属性值可以通过result()方法获取，这个方法是阻塞的，只有当线程将函数执行完并将函数结果通过set_result方法赋值给_result属性上时，result()方法才能得到返回结果，这里面的异步获取返回结果的过程是通过event()来控制的。
这种通过函数作为消息的方式来传递并由消费线程来执行的思想在一些分布式系统里面会经常用到。后面碰到这种案例再给兄弟们分享出来。