代码之家 › 专栏 › 技术社区 › systempuntoout

如何实现支持名称空间的FIFO队列

fifo queue google-app-engine python

2

systempuntoout · 技术社区 · 14 年前

我使用以下方法来处理基于Google应用引擎db.model的FIFO队列( see this question )

from google.appengine.ext import db
from google.appengine.ext import webapp
from google.appengine.ext.webapp import run_wsgi_app

class QueueItem(db.Model):
  created = db.DateTimeProperty(required=True, auto_now_add=True)
  data = db.BlobProperty(required=True)

  @staticmethod
  def push(data):
    """Add a new queue item."""
    return QueueItem(data=data).put()

  @staticmethod
  def pop():
    """Pop the oldest item off the queue."""
    def _tx_pop(candidate_key):
      # Try and grab the candidate key for ourselves. This will fail if
      # another task beat us to it.
      task = QueueItem.get(candidate_key)
      if task:
        task.delete()
      return task
    # Grab some tasks and try getting them until we find one that hasn't been
    # taken by someone else ahead of us
    while True:
      candidate_keys = QueueItem.all(keys_only=True).order('created').fetch(10)
      if not candidate_keys:
        # No tasks in queue
        return None
      for candidate_key in candidate_keys:
        task = db.run_in_transaction(_tx_pop, candidate_key)
        if task:
          return task

此队列按预期工作(非常好)。

现在,我的代码有一个方法可以访问这个由延迟队列调用的FIFO队列:

def deferred_worker():
        data= QueueItem.pop()
        do_something_with(data)

我想增强这个方法和队列数据结构,添加一个客户机ID参数来表示需要访问自己队列的特定客户机。类似:

def deferred_worker(client_ID):
        data= QueueItem_of_this_client_ID.pop() # I need to implement this
        do_something_with(data)

如何将队列编码为客户机识别?

制约因素:
-客户端的数量是动态的,并且不是预先定义的。
-taskqueue不是选项(1)。最多10个队列2。我想完全控制我的队列)

你知道我怎么用新的 Namespaces api (记得我不是从webapp.requesthandler调用db.model吗)?
另一个选择:我可以添加 client_ID db.StringProperty 对于使用它的queueitem,它有一个filter-on-pull方法:

QueueItem.all(keys_only=True).filter(client_ID=an_ID).order('created').fetch(10)

有更好的主意吗?

2 回复 | 直到 14 年前

1

Robert Kluin 14 年前

假设您的“客户机类”实际上是客户机调用的请求处理程序,您可以这样做:

from google.appengine.api import users
from google.appengine.api.namespace_manager import set_namespace

class ClientClass(webapp.RequestHandler):
  def get(self):
    # For this example let's assume the user_id is your unique id.
    # You could just as easily use a parameter you are passed.
    user = users.get_current_user()
    if user:
       # If there is a user, use their queue.  Otherwise the global queue.
       set_namespace(user.user_id())

    item = QueueItem.pop()
    self.response.out.write(str(item))

    QueueItem.push('The next task.')

或者,也可以设置命名空间 app-wide .

通过设置默认名称空间,对数据存储的所有调用都将“在”该名称空间内,除非您明确指定了其他名称空间。请注意,要获取和运行任务,您必须知道名称空间。因此,为了清理的目的,您可能需要维护默认名称空间中的名称空间列表。

2

1

Nick Johnson 14 年前

正如我在回答您对我原始答案的查询时所说的,您不需要对名称空间执行任何操作:构建队列的数据存储已经支持名称空间。只需根据需要设置命名空间,如中所述 the docs .