对象池设计模式的 Python 实现

Posted 2023-03-31

【中文标题】对象池设计模式的 Python 实现

【英文标题】：Python Implementation of the Object Pool Design Pattern

【发布时间】：2010-12-03 14:27:00

【问题描述】：

我需要一个Object Pool，而不是自己实现它，我想我会四处寻找一个现成且经过测试的 Python 库。

我发现很多其他人looking，但没有得到很多直接的答案，所以我把它带到了 Stack Overflow。

在我的例子中，我有大量线程（使用threading 模块），需要偶尔调用远程基于 SOAP 的服务器。他们每个人都可以建立自己与服务器的连接，但是设置一个套接字并完成身份验证过程很昂贵（它受到服务器的限制），所以我想共享一个连接池，只在需要时创建更多。

如果要池化的项目是工作子进程，我可能会选择multiprocessing.pool，但事实并非如此。如果它们是工作线程，我可能会选择this implementation，但它们不是。

如果它们是 mysql 连接，我可能会选择 pysqlpool，但事实并非如此。同样，SQLAlchemy Pool 已失效。

如果有一个线程，使用可变数量的连接/对象，我会考虑this implementation，但我需要它是线程安全的。

我知道我可以很快再次实现它，但鉴于有很多人在寻找它，我认为 Stack Overflow 上的规范答案会很好。

【参考方案1】：

在我看来，根据您的描述，您需要的是一个连接 池，而不是对象。为了简单的线程安全，只需将可重用连接保留在Queue.Queue 实例中，称为pool。当一个线程实例化一个连接包装对象时，该对象通过pool.get() 获得它的连接（如果当前没有可用的连接，它会自动将其入队等待，并在连接准备好时将其出列）；当对象完成使用其连接后，它会通过pool.put 将其放回池中。

除了Queue.Queue 已经为您提供的功能之外，其中几乎没有普遍需要的通用功能，因此没有模块提供众所周知或流行的功能也就不足为奇了——当一个模块有总共大约有 6 行功能代码（例如，调用用户提供的连接工厂以提前或及时填充队列直到某个最大数量——无论如何，这通常不是一个很大的附加值）。 “厚胶”，从标准库模块中厚厚地包装底层功能而没有实质性的附加值，毕竟是一种架构上的缺点；-)。

【讨论】：

啊，对了，等一下，如果池中没有更多内容，那就是列表缺少的内容。我以为我用列表而不是队列很聪明，但实际上太聪明了。 :)

@Lennart，也没有保证线程安全，您可能会也可能不会遇到问题，具体取决于实现——使用 Queue.Queue，您的线程安全得到了保证。

Python 已经内置了线程安全队列？我不知道！是的，这将加快实现速度（我认为这会很短，但主要花在思考并发问题上）。抱歉，我不明白您对“连接池”与“对象池”的区别。我说我想“共享一个连接池”，但是每个连接都包裹在一个对象中，所以它确实也是一个对象池。不过，我试图做出的区别是连接对象不是活动的（与 multiprocessing.pool 不同。）

@Oddthinking，是的，Python 标准库中的Queue 模块正是这样——一个线程安全队列（基础队列是 LIFO，也有优先级和 FIFO 变体）。至于“什么池”，我的观点是：池连接尽可能轻包装或展开，因为建立连接是昂贵的部分；将当前未使用的连接包装在一个全新的包装对象中，该对象在一个事务的持续时间内添加您想要的所有修剪，相比之下应该便宜且快速，因此无需将包装器池化！

【参考方案2】：

我遇到了类似的问题，我必须说 Queue.Queue 非常好，但是还缺少一块拼图。下面的类有助于确保获取的对象返回到池中。包括示例。

我允许使用 2 种方法来使用这个类，使用关键字或使用析构函数封装对象。 with 关键字是首选，但如果您由于某种原因不能/不想使用它（最常见的是需要来自多个队列的多个对象），至少您有一个选择。如果您选择使用该方法，则适用关于不调用析构函数的标准免责声明。

希望这可以帮助遇到与 OP 和我自己相同问题的人。

class qObj():
  _q = None
  o = None

  def __init__(self, dQ, autoGet = False):
      self._q = dQ

      if autoGet == True:
          self.o = self._q.get()

  def __enter__(self):
      if self.o == None:
          self.o = self._q.get()
          return self.o
      else:
          return self.o 

  def __exit__(self, type, value, traceback):
      if self.o != None:
          self._q.put(self.o)
          self.o = None

  def __del__(self):
      if self.o != None:
          self._q.put(self.o)
          self.o = None


if __name__ == "__main__":
  import Queue

  def testObj(Q):
      someObj = qObj(Q, True)

      print 'Inside func: 0'.format(someObj.o)

  aQ = Queue.Queue()

  aQ.put("yam")

  with qObj(aQ) as obj:
      print "Inside with: 0".format(obj)

  print 'Outside with: 0'.format(aQ.get())

  aQ.put("sam")

  testObj(aQ)

  print 'Outside func: 0'.format(aQ.get())

  '''
  Expected Output:
  Inside with: yam
  Outside with: yam
  Inside func: sam
  Outside func: sam
  '''

【讨论】：

恕我直言，唯一缺少的是 __getattr__ 方法，因此您可以将其视为封装对象的包装器

【参考方案3】：

对于简单的用例，这里是基于列表的对象池模式的示例实现：

来源：

https://sourcemaking.com/design_patterns/object_pool

https://sourcemaking.com/design_patterns/object_pool/python/1

"""
Offer a significant performance boost; it is most effective in
situations where the cost of initializing a class instance is high, the
rate of instantiation of a class is high, and the number of
instantiations in use at any one time is low.
"""


class ReusablePool:
    """
    Manage Reusable objects for use by Client objects.
    """

    def __init__(self, size):
        self._reusables = [Reusable() for _ in range(size)]

    def acquire(self):
        return self._reusables.pop()

    def release(self, reusable):
        self._reusables.append(reusable)


class Reusable:
    """
    Collaborate with other objects for a limited amount of time, then
    they are no longer needed for that collaboration.
    """

    pass


def main():
    reusable_pool = ReusablePool(10)
    reusable = reusable_pool.acquire()
    reusable_pool.release(reusable)


if __name__ == "__main__":
    main()

【讨论】：

考虑 Python 队列而不是 ReusablePool 中的列表，因为它被宣传为线程安全的。我不确定 Python 列表是否在所有实现中。