如何通过AMQP 1.0在Azure事件中心中批量接收许多消息

Question

I set up an AMQP 1.0 link with just the path and a filter using the Apache Qpid Electron Go wrapper for Qpid Proton like this:

amqpConnection.Receiver(
            // the path containing the consumer group
            // and the partition Id
            electron.Source("<EVENTHUB_PATH>"),
            // the filter map contains some annotations filters 
            // for the Event Hub offset
            electron.Filter(filterMap), 
)

I followed this doc to setup the AMQP link options: https://godoc.org/qpid.apache.org/electron#LinkOption

However while running the Go app I've realised it was very slow in fetching messages, so I've added 2 more link options like this:

amqpConnection.Receiver(
            electron.Source("<EVENTHUB_PATH>"),
            electron.Capacity(100),
            electron.Prefetch(true),
            electron.Filter(filterMap),
)

but after adding the capacity and the prefetch link options I don't see any improvement in the performance.

I keep receiving approximately 10 messages every ~5 seconds from 4 parallel links (one link per partition).

I've tried to run the app with the environment variable PN_TRACE_RAW=true for the verbose output from Qpid Proton (cf. this: https://qpid.apache.org/releases/qpid-proton-0.18.0/proton/c/api/group__transport.html), but I am not sure on what should I look for to troubleshoot this issue.

I don't think there is any issue with the Qpid settings, but anyway this is what I see on the terminal:

[0x9fd490]:0 -> @attach(18) [name="<MY_CUSTOM_NAME>",
handle=1, role=true, snd-settle-mode=0, rcv-settle-mode=0, source=@source(40) [address="<MY_CUSTOM_PATH>",
durable=0, expiry-policy=:"link-detach", timeout=0, dynamic=false, filter={:string=@:"apache.org:selector-filter:string"
"amqp.annotation.x-opt-offset > '<MY_CUSTOM_OFFSET>'"}], target=@target(41) [address="",
durable=0, expiry-policy=:"link-detach", timeout=0, dynamic=false], initial-delivery-count=0,
max-message-size=0]
[0x9fd490]:0 -> @flow(19) [next-incoming-id=1, incoming-window=2147483647, next-outgoing-id=1,
outgoing-window=0, handle=1, delivery-count=0, link-credit=100, drain=false]

I also tried to run the Go app in a Azure VM in the same Azure Location as the Event Hub, but no improvement in the performance.

How could I fetch many messages at the same time in the same "round trip"? I need to process thousands of messages per seconds.

Answer 1

You are correct that you need a prefetch window but the electron client can do a LOT better than that.

I did a quick test with the electron examples from https://github.com/apache/qpid-proton/tree/master/examples/go/electron

I get 3000 msg/sec even without prefetch, and nearly 10000 msgs/sec with.

$ ./broker -qsize 100000 &
Listening on [::]:5672

$  ./send -count 10000 /x ; time ./receive -count 10000 /x
Received all 10000 acknowledgements
Listening on 1 connections
Received 10000 messages

real    0m2.612s
user    0m1.611s
sys 0m0.510s

$  ./send -count 10000 /x ; time ./receive -count 10000 -prefetch 1000 /x
Received all 10000 acknowledgements
Listening on 1 connections
Received 10000 messages

real    0m1.053s
user    0m1.272s
sys 0m0.277s

There is clearly something funny going on - I'd like to help you get to the bottom of it.

PN_TRACE_RAW is a bit too verbose to be helpful, try PN_TRACE_FRM=1 which will give you a more readable summary.

I'm happy to continue the conversation either here or on [e-mail] if it turns into more of a support case than a question/answer.