nsq_tornado_asyncio_reader.py

import nsq
import asyncio
import tornado
import tornado.ioloop
import tornado.platform.asyncio

queue = asyncio.Queue()

async def go():
    def enqueue_message(message):
        message.enable_async()
        queue.put_nowait(message)

    reader = nsq.Reader(message_handler=enqueue_message,
        lookupd_http_addresses=['http://127.0.0.1:4161'],
        topic='test', channel='async')

    while True:
        message = await queue.get()
        print(message.body)
        message.finish()
        queue.task_done()

if __name__ == '__main__':
    tornado.platform.asyncio.AsyncIOMainLoop().install()
    loop = asyncio.get_event_loop()
    loop.run_until_complete(go())