京公网安备11010502063487号跨数据中心、跨集群搬消息有两个官方工具:Shovel 和 Federation。它们各是什么、怎么配、目的端挂了会怎样、消息是被复制还是被搬走?这篇起了两个独立的 broker,把两者各跑一遍。
$ rabbitmqctl set_parameter shovel my-shovel '{"src-queue":"src","dest-queue":"dst","ack-mode":"on-confirm",…}'
shovel_status: [{'name': 'my-shovel', 'type': 'dynamic', 'state': 'running', 'destination_protocol': 'amqp091', 'source_protocol': 'amqp091'}]
A.src 里剩 0 条,B.dst 里有 1000 条一个 shovel 就是“在源端当消费者、在目的端当发布者”的进程,用 rabbitmqctl set_parameter shovel 动态定义。ack-mode 决定何时确认源消息:on-confirm 最安全,no-ack 最快但可能丢。
on-confirm 905 ms 22106 条/秒 on-publish 974 ms 20533 条/秒 no-ack 874 ms 22870 条/秒
on-confirm(目的端确认后才确认源)、on-publish(发布后就确认)、no-ack(源端自动确认)三种吞吐几乎一样,约 2 万条/秒,差别在噪声内;越保守越不会丢。
B 宕机期间 shovel_status: [('my-shovel', 'terminated')];A.src 里积压 500 条
B 重新启动并可用之后 0.4 s,B.dst 里有 500 条,A.src 里剩 0 条目的端宕机时 shovel 终止,源队列保留消息;目的端回来后 shovel 重连、继续搬,按 reconnect-delay 重试。
$ rabbitmqctl set_parameter federation-upstream up-b '{"uri":"amqp://…B"}'
$ rabbitmqctl set_policy fed-x "^fx$" '{"federation-upstream-set":"all"}' --apply-to exchanges
B 上发布 5 条 orders.new + 1 条 other.x;A.fq_local(绑定 orders.#)里有 5 条
联邦链路状态(eval rabbit_federation_status:status()): [[{exchange,<<"fx">>}, {upstream_exchange,<<"fx">>}, {type,exchange}, {vhost,<<"/">>}, {upstream,<<"up-b">>}, {id,<<"c988aec5">>}, {status,running}, {local_connection,<<"<q30a@localhost.1791861923.3125.0>">>}, {uri,<<"amqp://127.0.0.1:9508">>},下游的联邦交换机按自己的绑定,向上游交换机拉取匹配的消息(复制),不匹配的不传。用 federation-upstream 参数 + policy 配置。
建好联邦链路、A 上还没有消费者:A.fqq 0 条,B.fqq 100 条(消息都还在上游) A 上开一个消费者后,0.0 s 收到 100 条;B.fqq 里剩 0 条(消息是被"搬"过去的,不是复制)
联邦队列只在下游有消费者时才从上游拉消息,是搬运不是复制;没有消费者时消息留在上游。