关于tornado+rabbitmq

接触到的文章:

• http://blog.csdn.net/weiyang7545/article/details/51742290
• https://github.com/geasyheart/swift_rpc
• https://github.com/zetaops/websocket-tornado-rabbitmq-example

今天开始测试发送工作..

rabbitmq pub:

import json
import logging

import pika
import tornado.ioloop
from tornado.web import Application
from tornado.websocket import WebSocketHandler

LOG_FORMAT = ('%(levelname) -10s %(asctime)s %(name) -30s %(funcName) '
              '-35s %(lineno) -5d: %(message)s')
LOGGER = logging.Logger(__name__)


class Publisher(object):
    EXCHANGE = 'logs'
    EXCHANGE_TYPE = 'fanout'
    ROUTING_KEY = ''

    def __init__(self, amqp_url):
        self._connection = None
        self._channel = None
        self._stopping = False
        self._url = amqp_url
        self._closing = False

    def connect(self):
        LOGGER.info('Connecting to %s', self._url)
        self._connection = pika.adapters.TornadoConnection(pika.URLParameters(self._url),
                                                           self.on_connection_open)

    def on_connection_open(self, unused_connection):
        LOGGER.info('Connection opened')
        self.open_channel()

    def open_channel(self):
        LOGGER.info('Creating a new channel')
        self._connection.channel(on_open_callback=self.on_channel_open)

    def on_channel_open(self, channel):
        LOGGER.info('Channel opened')
        self._channel = channel
        self.setup_exchange(self.EXCHANGE)

    def setup_exchange(self, exchange_name):
        LOGGER.info('Declaring exchange %s', exchange_name)
        self._channel.exchange_declare(self.on_exchange_declareok,
                                       exchange_name,
                                       self.EXCHANGE_TYPE)

    def on_exchange_declareok(self, unused_frame):
        LOGGER.info('Exchange declared')

    def publish_message(self, message):
        properties = pika.BasicProperties(delivery_mode=2, )

        self._channel.basic_publish(self.EXCHANGE, self.ROUTING_KEY,
                                    json.dumps(message, ensure_ascii=False),
                                    properties)

        LOGGER.info('Published message')


class IndexHandler(WebSocketHandler):
    links = []

    def open(self):
        self.links.append(self)

    def on_message(self, message):
        app.pika.publish_message(message)


app = Application([
    (r'/', IndexHandler)
])

url = "amqp://zy:123456@192.168.1.158:5672/%2F?connection_attempts=3&heartbeat_interval=3600"
app.pika = Publisher(url)
app.listen(80)
ioloop = tornado.ioloop.IOLoop.current()
ioloop.add_timeout(500, app.pika.connect)
ioloop.start()

赏

谢谢你请我吃糖果

支付宝