1

我基本上复制了这个例子。只是试图让最基本的工作：发布和消费。因此添加了一个简单的basic_publish。

消费者.py

# -*- coding:utf-8 -*-

import pika
from pika import exceptions
from pika.adapters import twisted_connection
from twisted.internet import defer, reactor, protocol, task
from datetime import datetime

print(datetime.now())
PAYLOAD = '{"foo": "bar"}'

connection_string = 'amqp://{}:{}@{}:5672/{}?backpressure_detection=t'.\
    format('scraper', 'password', 'rabbitmq_scraper', 'scraper')

@defer.inlineCallbacks
def run(connection):
    channel = yield connection.channel()
    exchange = yield channel.exchange_declare(exchange='topic_link', exchange_type='topic')
    queue = yield channel.queue_declare(queue='hello', auto_delete=False, exclusive=False)
    yield channel.queue_bind(exchange='topic_link', queue='hello', routing_key='hello.world')
    yield channel.basic_qos(prefetch_count=1)
    queue_object, consumer_tag = yield channel.basic_consume(queue='hello', no_ack=False)
    l = task.LoopingCall(read, queue_object)
    l.start(0.01)
    channel.basic_publish(exchange='topic_link', routing_key='hello.world', body=PAYLOAD, properties=None, mandatory=False, immediate=True)

@defer.inlineCallbacks
def read(queue_object):
    ch,method,properties,body = yield queue_object.get()
    if body:
        print(body)
    yield ch.basic_ack(delivery_tag=method.delivery_tag)

parameters = pika.URLParameters(connection_string)
cc = protocol.ClientCreator(reactor, twisted_connection.TwistedProtocolConnection, parameters)
d = cc.connectTCP('rabbitmq_scraper', 5672)
d.addCallback(lambda protocol: protocol.ready)
d.addCallback(run)
reactor.run()

执行与python3 -u consumer.py

我得到的唯一错误是：Unhandled error in Deferred:. 就这样？相当令人沮丧。

  1. 为什么会抛出异常？
  2. 我的追溯在哪里？
4

0 回答 0