Python定义一个Actor任务
作者:David Beazley 发布时间:2022-05-30 09:24:50
问题
你想定义跟actor模式中类似“actors”角色的任务
解决方案
actor模式是一种最古老的也是最简单的并行和分布式计算解决方案。 事实上,它天生的简单性是它如此受欢迎的重要原因之一。 简单来讲,一个actor就是一个并发执行的任务,只是简单的执行发送给它的消息任务。 响应这些消息时,它可能还会给其他actor发送更进一步的消息。 actor之间的通信是单向和异步的。因此,消息发送者不知道消息是什么时候被发送, 也不会接收到一个消息已被处理的回应或通知。
结合使用一个线程和一个队列可以很容易的定义actor,例如:
from queue import Queue
from threading import Thread, Event
# Sentinel used for shutdown
class ActorExit(Exception):
pass
class Actor:
def __init__(self):
self._mailbox = Queue()
def send(self, msg):
'''
Send a message to the actor
'''
self._mailbox.put(msg)
def recv(self):
'''
Receive an incoming message
'''
msg = self._mailbox.get()
if msg is ActorExit:
raise ActorExit()
return msg
def close(self):
'''
Close the actor, thus shutting it down
'''
self.send(ActorExit)
def start(self):
'''
Start concurrent execution
'''
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 method to be implemented by the user
'''
while True:
msg = self.recv()
# Sample ActorTask
class PrintActor(Actor):
def run(self):
while True:
msg = self.recv()
print('Got:', msg)
# Sample use
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()
讨论
actor模式的魅力就在于它的简单性。 实际上,这里仅仅只有一个核心操作 send()
. 甚至,对于在基于actor系统中的“消息”的泛化概念可以已多种方式被扩展。 例如,你可以以元组形式传递标签消息,让actor执行不同的操作,如下:
class TaggedActor(Actor):
def run(self):
while True:
tag, *payload = self.recv()
getattr(self,'do_'+tag)(*payload)
# Methods correponding to different message tags
def do_A(self, x):
print('Running A', x)
def do_B(self, x, y):
print('Running B', x, y)
# Example
a = TaggedActor()
a.start()
a.send(('A', 1)) # Invokes do_A(1)
a.send(('B', 2, 3)) # Invokes do_B(2,3)
a.close()
a.join()
作为另外一个例子,下面的actor允许在一个工作者中运行任意的函数, 并且通过一个特殊的Result对象返回结果:
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))
# Example use
worker = Worker()
worker.start()
r = worker.submit(pow, 2, 3)
worker.close()
worker.join()
print(r.result())
最后,“发送”一个任务消息的概念可以被扩展到多进程甚至是大型分布式系统中去。 例如,一个类actor对象的 send()
方法可以被编程让它能在一个套接字连接上传输数据 或通过某些消息中间件(比如AMQP、ZMQ等)来发送。
来源:https://python3-cookbook.readthedocs.io/zh_CN/latest/c12/p10_defining_an_actor_task.html
猜你喜欢
- 认知心理学家Donald A. Norman将人类的行动分解成七个阶段:1. 确定目标2. 确定意图3. 明确行动内容4. 执行5. 感知外
- 放大镜并不是一个难以实现的效果, 只是因为牵涉到一些精确的数值计算, 显得比较繁琐. 在未来的一段日子, 我会不定期地写关于 JavaScr
- 大家好,今天在写代码的时候,遇到了这样一种情况。我有如下所示的几个类用来存放程序配置(其实当做命名空间来用,同时感觉能够继承方便一点),im
- 1,file_put_contents()函数 2,使用PHP内置缓存机制实现页面静态化:output_bufferingphp中outpu
- 本文实例讲述了Python3.5内置模块之time与datetime模块用法。分享给大家供大家参考,具体如下:1、模块的分类a、标准库(Py
- 代码如下:Function htmll(mulu,htmlmulu,FileName,filefrom,htmla,htmlb,h
- 好几年前我在做一些自动化的脚本时,脑子里也闪过这样的想法:能不能直接把浏览器的cookies取出来用呢?直到昨天看到代码《python模拟发
- 因为要批量用某软件处理一批eps文件,所以要模拟鼠标及键盘动作,使其能够自动化操作。#-*-coding:utf-8-*-import os
- 1. 首先是环境的安装 (本人使用的是PyCharm,python3.6)pip3 install PyQt5 (没有指定版本的话,默认会安
- 一、报错: 「Can't swap PDO instance while within transaction」通过查询 Larav
- 前言我们的游戏资源处理工具是Python实现的,功能包括csv解析,UI材质处理,动画资源解析、批处理,Androd&iOS自动打包
- 在开发中有些敏感接口,例如用户余额提现接口,需要考虑在并 * 况下接口是否会发生问题。如果用户将自己的多条提现请求同时发送到服务器,代码能否扛
- 如何在ADO服务器端利用好缓存技术?请看下面示例,这是一个用来显示图书分类的例子程序:displayBooks.asp< %
- 一直在用JS写ASP,也不是特别原因,只是当初学的是JS,后来学ASP时知道ASP也可以用JS写,就没去学VBS.前几个月刚学ASP的时候找
- 本文实例讲述了Python简单计算文件MD5值的方法。分享给大家供大家参考,具体如下:一 代码import sysimport hashli
- 当Python执行import语句时,它会在一些路径中搜索Python模块和扩展模块。可以通过sys.path查看这些路径,比如:>&
- http://pyhdfs.readthedocs.io/en/latest/1:安装由于是windows环境(linux其实也一样),只要
- Dean Edwards 最近有篇文章很精彩,忍不住在这里翻译下。-- Split --很多 Javascript 框架都提供了自定义事件(
- 今天跟大家分享一下,如何通过Python实现一个轻量级的库来获取电脑上连接的Android设备信息,为什么说轻量呢因为整个库也就4KB,相比
- 前言MERGE语句是Oracle9i新增的语法,用来合并UPDATE和INSERT语句。 通过MERGE语句,根据一张表或多表联合查询的连接