問題

你想定義跟actor模式中類似“actors”角色的任務(wù)

解決方案

actor模式是一種最古老的也是最簡單的并行和分布式計算解決方案。 事實上，它天生的簡單性是它如此受歡迎的重要原因之一。 簡單來講，一個actor就是一個并發(fā)執(zhí)行的任務(wù)，只是簡單的執(zhí)行發(fā)送給它的消息任務(wù)。 響應(yīng)這些消息時，它可能還會給其他actor發(fā)送更進一步的消息。 actor之間的通信是單向和異步的。因此，消息發(fā)送者不知道消息是什么時候被發(fā)送， 也不會接收到一個消息已被處理的回應(yīng)或通知。

結(jié)合使用一個線程和一個隊列可以很容易的定義actor，例如：

from queue import Queuefrom threading import Thread, Event# Sentinel used for shutdownclass ActorExit(Exception): passclass 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 ActorTaskclass PrintActor(Actor): def run(self): while True: msg = self.recv() print(’Got:’, msg)# Sample usep = PrintActor()p.start()p.send(’Hello’)p.send(’World’)p.close()p.join()

這個例子中，你使用actor實例的 send() 方法發(fā)送消息給它們。 其機制是，這個方法會將消息放入一個隊里中， 然后將其轉(zhuǎn)交給處理被接受消息的一個內(nèi)部線程。 close() 方法通過在隊列中放入一個特殊的哨兵值（ActorExit）來關(guān)閉這個actor。 用戶可以通過繼承Actor并定義實現(xiàn)自己處理邏輯run()方法來定義新的actor。 ActorExit 異常的使用就是用戶自定義代碼可以在需要的時候來捕獲終止請求 （異常被get()方法拋出并傳播出去）。

如果你放寬對于同步和異步消息發(fā)送的要求， 類actor對象還可以通過生成器來簡化定義。例如：

def print_actor(): while True: try: msg = yield # Get a message print(’Got:’, msg) except GeneratorExit: print(’Actor terminating’)# Sample usep = print_actor()next(p) # Advance to the yield (ready to receive)p.send(’Hello’)p.send(’World’)p.close()

討論

actor模式的魅力就在于它的簡單性。 實際上，這里僅僅只有一個核心操作 send() . 甚至，對于在基于actor系統(tǒng)中的“消息”的泛化概念可以已多種方式被擴展。 例如，你可以以元組形式傳遞標簽消息，讓actor執(zhí)行不同的操作，如下：

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)# Examplea = 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允許在一個工作者中運行任意的函數(shù)， 并且通過一個特殊的Result對象返回結(jié)果：

from threading import Eventclass 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._resultclass 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 useworker = Worker()worker.start()r = worker.submit(pow, 2, 3)worker.close()worker.join()print(r.result())

最后，“發(fā)送”一個任務(wù)消息的概念可以被擴展到多進程甚至是大型分布式系統(tǒng)中去。 例如，一個類actor對象的 send() 方法可以被編程讓它能在一個套接字連接上傳輸數(shù)據(jù) 或通過某些消息中間件（比如AMQP、ZMQ等）來發(fā)送。

以上就是Python定義一個Actor任務(wù)的詳細內(nèi)容，更多關(guān)于Python actor任務(wù)的資料請關(guān)注好吧啦網(wǎng)其它相關(guān)文章！