Tornado 和Redis的新手，并实现了侦听器/工作程序设置的开始。
我希望能够将任务推送到队列中，然后将它们弹出。 BRPOP似乎是将它们弹出的最佳方法，因为如果当前没有它们，它会等待添加一个。问题是，每当我使用它时，它就永远不会返回...但是，当我使用RPOP时，我按预期得到了队列中的下一个项目。
class ListenHandler(tornado.websocket.WebSocketHandler):
uid = ''
CHANNEL_TPL = "client_%s"
RESPONSE_TPL = '{"command":"%s","rid":"%s","status":"%s","result":%s}'
def open(self):
# new websocket connection is established from a client
print "open socket"
self.uid = session = uuid4()
def on_message(self, message):
print "on_message called [%s]" % message
try:
m = json.loads(message)
except ValueError:
self.write_message('BAD')
return
# check for RID (request id)
if not 'rid' in m:
self.write_message('error: unspecified rid')
return
# confirm receipt of data
confirm_string = '%s OK' % (m['rid'])
self.write_message(confirm_string)
# check for command
if not 'command' in m:
response = '%s error: unspecified command' % (m['rid'])
self.write_message(response)
return
# process commands
if m['command'] == 'register':
self._register(m['rid'])
elif m['command'] == 'get_canvas':
self._queue_command('read', m)
elif m['command'] == 'save_canvas':
if 'data' in m:
self._queue_command('write', m)
else:
response = '%s unspecified data' % (m['rid'])
self.write_message(response)
return
elif m['command'] == 'list_command_queue':
self._list_command_queue(m['rid'])
elif m['command'] == 'get_read_job':
self._get_read_job(m['rid'])
elif m['command'] == 'get_write_job':
self._get_write_job(m['rid'])
else:
# no commands recognized
response = '%s error: unknown command' % (m['rid'])
self.write_message(response)
return
print "end of on_message()\n"
def callback(self, data):
print "- callback()"
self.write_message(data)
def on_close(self):
# websocket connection is closed by client
pass
def _register(self, rid):
data = '{"uid":"%s"}' % (self.uid)
response = self.RESPONSE_TPL % ('register', rid, 'completed', data)
self.callback(response)
@tornado.web.asynchronous
@tornado.gen.engine
def _queue_command(self, type, m):
channel = self.CHANNEL_TPL % (type)
print "pushing job to %s ... data[%s]" % (channel, m)
yield tornado.gen.Task(self.application.rdb.lpush, channel, m)
return
def _list_command_queue(self, rid):
channel_r = self.CHANNEL_TPL % ('read')
channel_w = self.CHANNEL_TPL % ('write')
data = '{"client_read":"%s","client_write":"%s"}' % (self.application.rdb.llen(channel_r), self.application.rdb.llen(channel_w))
response = self.RESPONSE_TPL % ('list_command_queue', rid, 'completed', data)
print "list_command_queue [%s]" % (response)
self.callback(response)
@tornado.web.asynchronous
@tornado.gen.engine
def _get_read_job(self, rid):
channel = self.CHANNEL_TPL % ('read')
data = yield tornado.gen.Task(self.application.rdb.rpop, (channel))
response = self.RESPONSE_TPL % ('get_read_job', rid, 'completed', data)
self.callback(response)
@tornado.web.asynchronous
@tornado.gen.engine
def _get_write_job(self, rid):
channel = self.CHANNEL_TPL % ('write')
data = yield tornado.gen.Task(self.application.rdb.rpop, (channel))
response = self.RESPONSE_TPL % ('get_write_job', rid, 'completed', data)
self.callback(response)
上面的类将接受的命令并将它们LPUSH放入两个不同的队列之一(用于将写入SQL DB的作业的“写”队列，以及用于只读作业的“读”队列)。理论上，使用BRPOP的另一台计算机上将有许多“ worker ”线程来获取这些命令并执行它们。不过，到目前为止，我正在使用相同的侦听器来测试队列中的内容。
函数get_write_jobs和get_read_jobs将返回队列中的下一个条目，没有问题。但是，如果我在任一方法中添加“b”(brpop)，该函数将永远无法调用回调。似乎只是锁定，等待下一个可用的条目，但是那里有条目。
知道这里发生了什么吗？我是否误解了BRPOP的目的？
谢谢，
缺口
最佳答案
也许有点晚了，但我想我可以帮忙。
我今天也遇到过同样的问题，这几乎使我发疯。最后，经过大量调试后，我通过将键或通道传递给列表(例如[channel])中的blpop / brpop来解决了该问题。
Tornado-redis的brpop / blpop内部执行的操作是将键转换为列表，但是当仅接收到一个键时，会将字符串转换为字符列表(简直令人惊讶...)，这就是为什么调用之后会阻塞的原因，它正在等待各种列表中的新项，这些项的名称与原始键的所有字符相对应。