我没有尝试使用txredis(redis的非阻塞扭曲api)作为持久消息队列,我正在尝试使用我正在处理的scrapy项目进行设置.我发现虽然客户端没有阻塞,但它变得比它本来要慢得多,因为应该在反应器循环中的一个事件被分成数千个步骤.
所以相反,我尝试使用redis-py(常规阻塞扭曲api)并将调用包装在延迟线程中.它工作得很好,但是当我打电话给redis时我想要执行内部延迟,因为我想设置连接池以试图进一步加快速度.
下面是我对一些延迟线程的扭曲文档中的一些示例代码的解释,以说明我的用例:
#!/usr/bin/env python from twisted.internet import reactor,threads from twisted.internet.task import LoopingCall import time def main_loop(): print 'doing stuff in main loop.. do not block me!' def aBlockingRedisCall(): print 'doing lookup... this may take a while' time.sleep(10) return 'results from redis' def result(res): print res def main(): lc = LoopingCall(main_loop) lc.start(2) d = threads.deferToThread(aBlockingRedisCall) d.addCallback(result) reactor.run() if __name__=='__main__': main()
这是我对连接池的更改,使得延迟线程中的代码阻塞:
#!/usr/bin/env python from twisted.internet import reactor,defer from twisted.internet.task import LoopingCall import time def main_loop(): print 'doing stuff in main loop.. do not block me!' def aBlockingRedisCall(x): if x<5: #all connections are busy, try later print '%s is less than 5, get a redis client later' % x x+=1 d = defer.Deferred() d.addCallback(aBlockingRedisCall) reactor.callLater(1.0,d.callback,x) return d else: print 'got a redis client; doing lookup.. this may take a while' time.sleep(10) # this is now blocking.. any ideas? d = defer.Deferred() d.addCallback(gotFinalResult) d.callback(x) return d def gotFinalResult(x): return 'final result is %s' % x def result(res): print res def aBlockingMethod(): print 'going to sleep...' time.sleep(10) print 'woke up' def main(): lc = LoopingCall(main_loop) lc.start(2) d = defer.Deferred() d.addCallback(aBlockingRedisCall) d.addCallback(result) reactor.callInThread(d.callback, 1) reactor.run() if __name__=='__main__': main()
所以我的问题是,有谁知道为什么我的更改导致延迟线程被阻止和/或任何人都可以提出更好的解决方案?
好吧,正如扭曲的文档所说:
延迟不会使代码神奇地不阻止
无论何时使用阻塞代码,例如sleep
,您都必须将其推迟到新线程.
#!/usr/bin/env python from twisted.internet import reactor,defer, threads from twisted.internet.task import LoopingCall import time def main_loop(): print 'doing stuff in main loop.. do not block me!' def aBlockingRedisCall(x): if x<5: #all connections are busy, try later print '%s is less than 5, get a redis client later' % x x+=1 d = defer.Deferred() d.addCallback(aBlockingRedisCall) reactor.callLater(1.0,d.callback,x) return d else: print 'got a redis client; doing lookup.. this may take a while' def getstuff( x ): time.sleep(3) return "stuff is %s" % x # getstuff is blocking, so you need to push it to a new thread d = threads.deferToThread(getstuff, x) d.addCallback(gotFinalResult) return d def gotFinalResult(x): return 'final result is %s' % x def result(res): print res def aBlockingMethod(): print 'going to sleep...' time.sleep(10) print 'woke up' def main(): lc = LoopingCall(main_loop) lc.start(2) d = defer.Deferred() d.addCallback(aBlockingRedisCall) d.addCallback(result) reactor.callInThread(d.callback, 1) reactor.run() if __name__=='__main__': main()
如果redis api不是很复杂,使用twisted.web重写它可能更自然,而不是仅仅在很多线程中调用阻塞api.