0

我编写了一个多生产者单消费者系统，其中Service消费来自不同线程的事件并处理它们。这是消费者的外观：

@dataclass
class RPCEvent:
    method: str
    args: Tuple[Any] = field(default_factory=tuple)
    kwargs: Dict[Any, Any] = field(default_factory=dict)

RPCEventQueue = Queue[RPCEvent] 

class Service:
    def __init__(self):
        self.queue = RPCEventQueue()
    
    def fun(self, message: str):
        print(f"fun({message})")

    def run(self):
        while True:
            try:
                event = self.queue.get(1)
                getattr(event.method)(*event.args, **event.kwargs)
            except Empty:
                continue
            except Exception:
                break
            finally:
                self.queue.task_done()

     # from another thread I push events into Service's queue
     # service.queue.put_nowait(RPCEvent(method="fun", kwargs={"message": "one"}))

这很好用，但是 和 之间没有语法可执行RPCEvent性Service。例如我可以这样做：

service.queue.put_nowait(RPCEvent(method="fun", kwargs={"oops": "oh no"}))

有没有办法知道在写这一行时Service.fun没有参数？oops

我检查了Protocol文档，但找不到适合我的情况的创造性方法。

4

0 回答 0