pydispatcher是否在后台线程中运行处理程序函数？

Question

在查找事件处理程序模块时，我遇到了pydispatcher，它似乎对初学者很友好。我对这个库的用例是，如果我的队列大小超过阈值，我想发送一个信号。然后，处理程序函数可以开始处理和删除队列中的项(随后在数据库中执行大容量插入)。

我希望处理函数在后台运行。我知道我可以简单地重写queue.append()方法，检查队列大小并异步调用处理函数，但我希望实现侦听器-分派器模型，以保持逻辑的整洁和分离。

pydispatcher可以开箱即用吗？如果没有，有没有其他模块可以帮我做到这一点？我是否需要管理对队列的访问，因为可能会有多个线程同时处理并追加到队列中？

请注意，在我的用例中，只有一个调度程序和事件处理程序。

Answer 1

我最近发布了Akuanduba模块，它可能会帮助您完成此任务。在存储库上有一个示例，可以帮助您了解它是如何工作的，它看起来与您想要的相似。

无论如何，我将尝试在这里解释一种使用Akuanduba实现代码的方法：

• 首先你可以创建一个数据框来保存你的队列：

# Mandatory imports
from Akuanduba.core.messenger.macros import *
from Akuanduba.core.constants import *
from Akuanduba.core import NotSet, AkuandubaDataframe
# Your imports go here:
from queue import Queue

class MyQueue (AkuandubaDataframe):

  def __init__(self, name):

    # Mandatory stuff
    AkuandubaDataframe.__init__(self, name)

    self.__queue = Queue ()

  def getQueue (self):
    return self.__queue

  def putQueue (self, val):
    self.__queue.put(val)

  def getQueueSize (self):
    return self.__queue.qsize()

  #
  # "toRawObj" method is a mandatory method that delivers a dict with the desired data
  # for file saving
  #
  def toRawObj(self):
    d = {
          "Queue" : self.getQueue(),
          }
    return d

• ，然后您可以创建一个检查队列大小的TriggerCondition：

from Akuanduba.core import StatusCode, NotSet, StatusTrigger
from Akuanduba.core.messenger.macros import *
from Akuanduba.core import TriggerCondition
import time

class CheckQueueSize (TriggerCondition):

  def __init__(self, name, maxSize):

    TriggerCondition.__init__(self, name)
    self._name = name
    self._maxSize = maxSize

  def initialize(self):

    return StatusCode.SUCCESS

  def execute (self):

    size = self.getContext().getHandler("MyQueue").getQueueSize()
    if (size > SIZE_THRESHOLD):
      return StatusTrigger.TRIGGERED
    else:
      return StatusTrigger.NOT_TRIGGERED

  def finalize(self):

    return StatusCode.SUCCESS

• 创建一个可用作处理函数的工具：

# Mandatory imports
from Akuanduba.core import AkuandubaTool, StatusCode, NotSet, retrieve_kw
# Your imports go here:

class SampleTool(AkuandubaTool):

  def __init__(self, name, **kw):

    # Mandatory stuff
    AkuandubaTool.__init__(self, name)


  def initialize(self):

    # Lock the initialization. After that, this tool can not be initialized once again
    self.init_lock()
    return StatusCode.SUCCESS


  def execute(self,context):

    #
    # DO SOMETHING HERE
    #

    # Always return SUCCESS
    return StatusCode.SUCCESS

  def finalize(self):
    self.fina_lock()
    return StatusCode.SUCCESS

• ，最后，创建一个主脚本，以使所有内容都能协同工作：

# Akuanduba imports
from Akuanduba.core import Akuanduba, LoggingLevel, AkuandubaTrigger
from Akuanduba import ServiceManager, ToolManager, DataframeManager

# This sample's imports
import MyQueue, CheckQueueSize, SampleTool

# Creating your handler
your_handler = SampleTool ("Your Handler's name")

# Creating dataframes
queue = MyQueue ("MyQueue")

# Creating trigger
trigger  = AkuandubaTrigger("Sample Trigger Name", triggerType = 'or')

# Append conditions and tools to trigger just adding them
# Tools appended to the trigger will only run when trigger is StatusTrigger.TRIGGERED,
# and will run in the order they've been appended
trigger += CheckQueueSize( "CheckQueueSize condition", MAX_QUEUE_SIZE )
trigger += your_handler

# Creating Akuanduba
manager = Akuanduba("Akuanduba", level=LoggingLevel.INFO)

# Appending tools
#
# ToolManager += TOOL_1
# ToolManager += TOOL_2
#
ToolManager += trigger

# Apprending dataframes
DataframeManager += sampleDataframe

# Initializing 
manager.initialize()
manager.execute()
manager.finalize()

这样，你就有了干净和分离的代码。