0

我使用 pika 与 rabbitmq 通信。我消耗一个队列，当收到消息时，我启动一个 pykka 的 actor 来运行我的逻辑，有时我的逻辑可能使用 pika 连接来发布消息。但我发现它会花费 5 秒在actor中创建一个频道。代码如下：

def on_conn_open(connection):
    connection.channel(lambda ch:ch.basic_consume(on_message,"q1"))
def on_message(channel, basic_deliver, properties, body):
    channel.basic_ack(basic_deliver.delivery_tag)
    body=body.decode(encoding = 'utf8')
    print(channel,body)
    if body=="go":
        log.debug("start...")
        actor.tell({})
conn=pika.SelectConnection(pika.URLParameters('amqp://guest:guest@127.0.0.1:5672'),on_open_callback=on_conn_open,stop_ioloop_on_close=False) 

class TTT(pykka.ThreadingActor):
    def on_receive(self, message):
        conn.channel(lambda ch:log.debug("ok"))

actor=TTT.start()
conn.ioloop.start()

这是控制台：

2016-11-11 15:04:52,292 test        : DEBUG    start...
2016-11-11 15:04:52,292 pika.connection: DEBUG    Creating channel 3
2016-11-11 15:04:52,292 pika.callback: DEBUG    Added: {'only': <Channel number=3 CLOSED conn=<SelectConnection OPEN socket=('127.0.0.1', 2314)->('127.0.0.1', 5672) params=<URLParameters host=127.0.0.1 port=5672 virtual_host=/ ssl=False>>>, 'calls': 1, 'one_shot': True, 'callback': <bound method Connection._on_channel_cleanup of <SelectConnection OPEN socket=('127.0.0.1', 2314)->('127.0.0.1', 5672) params=<URLParameters host=127.0.0.1 port=5672 virtual_host=/ ssl=False>>>, 'arguments': None}
2016-11-11 15:04:52,292 pika.callback: DEBUG    Added: {'only': None, 'one_shot': False, 'callback': <bound method Channel._on_getempty of <Channel number=3 OPENING conn=<SelectConnection OPEN socket=('127.0.0.1', 2314)->('127.0.0.1', 5672) params=<URLParameters host=127.0.0.1 port=5672 virtual_host=/ ssl=False>>>>, 'arguments': None}
2016-11-11 15:04:52,292 pika.callback: DEBUG    Added: {'only': None, 'one_shot': False, 'callback': <bound method Channel._on_cancel of <Channel number=3 OPENING conn=<SelectConnection OPEN socket=('127.0.0.1', 2314)->('127.0.0.1', 5672) params=<URLParameters host=127.0.0.1 port=5672 virtual_host=/ ssl=False>>>>, 'arguments': None}
2016-11-11 15:04:52,292 pika.callback: DEBUG    Added: {'only': None, 'one_shot': False, 'callback': <bound method Channel._on_flow of <Channel number=3 OPENING conn=<SelectConnection OPEN socket=('127.0.0.1', 2314)->('127.0.0.1', 5672) params=<URLParameters host=127.0.0.1 port=5672 virtual_host=/ ssl=False>>>>, 'arguments': None}
2016-11-11 15:04:52,292 pika.callback: DEBUG    Added: {'only': None, 'calls': 1, 'one_shot': True, 'callback': <bound method Channel._on_close of <Channel number=3 OPENING conn=<SelectConnection OPEN socket=('127.0.0.1', 2314)->('127.0.0.1', 5672) params=<URLParameters host=127.0.0.1 port=5672 virtual_host=/ ssl=False>>>>, 'arguments': None}
2016-11-11 15:04:52,292 pika.channel: DEBUG    Entering blocking state on frame <Channel.Open(['out_of_band='])>; acceptable_replies=[<class 'pika.spec.Channel.OpenOk'>]
2016-11-11 15:04:52,292 pika.channel: DEBUG    Adding on_synchronous_complete callback
2016-11-11 15:04:52,292 pika.callback: DEBUG    Added: {'only': None, 'calls': 1, 'one_shot': True, 'callback': <bound method Channel._on_synchronous_complete of <Channel number=3 OPENING conn=<SelectConnection OPEN socket=('127.0.0.1', 2314)->('127.0.0.1', 5672) params=<URLParameters host=127.0.0.1 port=5672 virtual_host=/ ssl=False>>>>, 'arguments': None}
2016-11-11 15:04:52,292 pika.channel: DEBUG    Adding passed-in callback
2016-11-11 15:04:52,292 pika.callback: DEBUG    Added: {'only': None, 'calls': 1, 'one_shot': True, 'callback': <bound method Channel._on_openok of <Channel number=3 OPENING conn=<SelectConnection OPEN socket=('127.0.0.1', 2314)->('127.0.0.1', 5672) params=<URLParameters host=127.0.0.1 port=5672 virtual_host=/ ssl=False>>>>, 'arguments': None}
2016-11-11 15:04:57,298 pika.callback: DEBUG    Processing 3:Channel.OpenOk
2016-11-11 15:04:57,298 pika.callback: DEBUG    Processing use of oneshot callback
2016-11-11 15:04:57,298 pika.callback: DEBUG    0 registered uses left
2016-11-11 15:04:57,298 pika.callback: DEBUG    Removing callback #0: {'only': None, 'calls': 0, 'one_shot': True, 'callback': <bound method Channel._on_synchronous_complete of <Channel number=3 OPENING conn=<SelectConnection OPEN socket=('127.0.0.1', 2314)->('127.0.0.1', 5672) params=<URLParameters host=127.0.0.1 port=5672 virtual_host=/ ssl=False>>>>, 'arguments': None}
2016-11-11 15:04:57,298 pika.callback: DEBUG    Processing use of oneshot callback
2016-11-11 15:04:57,298 pika.callback: DEBUG    0 registered uses left
2016-11-11 15:04:57,298 pika.callback: DEBUG    Removing callback #0: {'only': None, 'calls': 0, 'one_shot': True, 'callback': <bound method Channel._on_openok of <Channel number=3 OPENING conn=<SelectConnection OPEN socket=('127.0.0.1', 2314)->('127.0.0.1', 5672) params=<URLParameters host=127.0.0.1 port=5672 virtual_host=/ ssl=False>>>>, 'arguments': None}
2016-11-11 15:04:57,298 pika.callback: DEBUG    Calling <bound method Channel._on_synchronous_complete of <Channel number=3 OPENING conn=<SelectConnection OPEN socket=('127.0.0.1', 2314)->('127.0.0.1', 5672) params=<URLParameters host=127.0.0.1 port=5672 virtual_host=/ ssl=False>>>> for "3:Channel.OpenOk"
2016-11-11 15:04:57,298 pika.channel: DEBUG    0 blocked frames
2016-11-11 15:04:57,298 pika.callback: DEBUG    Calling <bound method Channel._on_openok of <Channel number=3 OPENING conn=<SelectConnection OPEN socket=('127.0.0.1', 2314)->('127.0.0.1', 5672) params=<URLParameters host=127.0.0.1 port=5672 virtual_host=/ ssl=False>>>> for "3:Channel.OpenOk"
2016-11-11 15:04:57,298 test        : DEBUG    ok

为什么 pika 在 pykka 中使用时创建频道总是需要 5 秒？请帮我解决这个问题

4

1 回答 1

0

看起来 pika 并不是真的适合线程，所以pika+pykka没有问题，但是对于pika+threads，一般来说，尝试使用GeventActor或EventletActor代替ThreadActor，你会看到一些改进。

来自 pika 的 github：

由于线程并不适用于所有情况，因此它不需要线程。小心不要禁止他们。greenlets、回调、延续和生成器也是如此。但是，它不一定是线程安全的，并且您的里程会有所不同。

于 2016-12-13T21:33:36.923 回答