戏剧演员中的无限异步循环

Question

我想在演剧演员内部进行无限循环（对于某些连续过程）。

我试图在receiveMessage_Run()方法中启动循环（参见下面的代码），但无法结束循环。参与者被阻止接收新消息

有人能帮我完成这项任务吗？ 如何运行无限循环（似乎是异步的）并保存对其他消息做出反应的可能性以及停止循环和参与者的可能性

我的测试代码：

from thespian.actors import ActorExitRequest, ActorSystem, Actor, ActorTypeDispatcher
import time
import random

# Message 'Run'
class Run():
    pass

# Message 'Stop'
class Stop():
    pass


class Actor(ActorTypeDispatcher):
    def __init__(self):
        self.stop = False

    # It seems, it must be asynchronous
    def loop(self):
        while not self.stop:
            time.sleep(1)
            print(random.random())

    def receiveMsg_Run(self, data, sender):
        self.loop()
        self.send(sender, 'Finished the loop.')

    def receiveMsg_Stop(self, msg, sender):
        self.stop = True
        self.send(self.myAddress, ActorExitRequest)
        print('*' * 1000)

    def receiveMsg_ActorExitRequest(self, msg, sender):
        self.stop = True
        time.sleep(1)


if __name__ == '__main__':
    asys = ActorSystem('multiprocQueueBase')

    loop = asys.createActor(Actor)

    resp = asys.ask(loop, Run())
    print(resp)
    time.sleep(4)
    asys.tell(loop, Stop())

    ActorSystem().shutdown()

Answer 1

该程序基本上在一个无限循环中运行每个actor，直到收到ActorExitRequest()或actor系统关闭。但是，每个actor只是一个执行线程，因此正如您所发现的，如果actor的receiveMessage()方法从未退出，则该程序无法运行以处理其他传入请求（或完成异步出站发送）。执行单线程的优点是，参与者的上下文是简单明了的：它不需要对成员变量进行互斥保护，或者考虑数据竞争、死锁或值损坏。p>

class Actor(ActorTypeDispatcher):
    def __init__(self):
        self.stop = False
        self.requestor = None

    def run_single(self):
            print(random.random())

    def receiveMsg_Run(self, data, sender):
        self.requestor = sender
        self.run_single()
        self.wakeupAfter(datetime.timedelta(seconds=1))

    def receiveMsg_WakeupMessage(self, msg, sender):
        if self.stop:
            self.send(self.requestor, 'Finished the loop.')
        else:
            self.run_single()
            self.wakeupAfter(datetime.timedelta(seconds=1))

    def receiveMsg_Stop(self, msg, sender):
        self.stop = True
        # Note that this next line will probably remove the need for self.stop entirely:
        # receiveMsg_WakeupMessage will not get a chance to run when self.stop is True
        self.send(self.myAddress, ActorExitRequest)
        print('*' * 1000)

    def receiveMsg_ActorExitRequest(self, msg, sender):
        self.stop = True
        time.sleep(1)

有关更多详细信息，请参见https://thespianpy.com/doc/using.html#hH-9cc1acd4-7f35-44ac-94fc-0326dc87a7a5

如果您的^ {< CD3>}确实阻塞了很长一段时间，不能通过上述方法定期调度，则可以考虑使用单独的线程；这是不同处理和调度概念（Actor模型v.s.线程）的不幸混合，但有时在Actor必须运行用该替代模型编写的代码时是必要的