TGIP-CN 024: AMQP on Pulsar 的设计与应用

1 .基于Pulsar架构的移动云AMQP消息队列 设计与应用 中国移动云能力中心 张浩 

2 . 目录 1 相关背景 2 AMQP 0-9-1 目录 3 Protocol Handler 4 我们怎么实现AoP 5 怎么使用AoP 6 怎么使用移动云上的AMQP 7 后续计划 2 

3 .相关背景 3 

4 .相关背景 • 我们为什么要自己做AMQP消息队列？ – OpenStack中默认使用Rabbitmq作为RPC通信组件 • 线上环境经常出问题，运维比较难 – 移动云客户的需求 • 移动云我们已上线消息队列RocketMQ以及MQTT服务，但是有不少 客户对AMQP消息队列需求强烈 • 我们为什么选择基于Pulsar来实现AMQP协议？ – 与StreamNative的合作开发 • 社区支持，共同开发AMQP on Pulsar. 4 

5 .AMQP 0-9-1 5 

6 .AMQP-0-9-1 什么是 AMQP 0-9-1? AMQP 0-9-1 (Advanced Message Queuing Protocol) is a messaging protocol that enables conforming client applications to communicate with conforming messaging middleware brokers. 6 

7 .AMQP-0-9-1 – VirtualHost • 资源的逻辑分组以及隔离 – Exchange • 消息路由 – Queue • 消息存储 – Binding • 路由规则 7 

8 .AMQP-0-9-1 “Hello, world” example routing 8 

9 .Protocol Handler 9 

10 .Pulsar Protocol Handler 10 

11 .Protocol Handlers 11 

12 .AMQP-on-Pulsar Protocol Handler 12 

13 .我们是怎么实现AoP? 13 

14 .实现 – Pulsar to AMQP model – Message Publish – Message Consume – Proxy 14 

15 .Pulsar to AMQP model AMQP Exchange-> Pulsar topic + Replicator 15 

16 .Pulsar to AMQP model AMQP Queue-> Pulsar topic + Router 16 

17 .Pulsar to AMQP model AMQP VirtualHost -> Pulsar tenant/namespace persistent :// my-tenant / my-namespace / my-topic Topic Domain Tenant Namespace Topic 17 

18 .Message Publish 18 

19 .Message Publish – AMQP message -> Pulsar message – topic.publishMessage – Exchange topic replicator -> Queue topic 19 

20 .Message Publish Exchange topic中为什么要添加replicator? 20 

21 .Message Consume Pulsar shared subscription-> AMQP default 21 

22 .Message Consume Pulsar exclusive subscription -> AMQP exclusive queue 22 

23 .Message Consume Pulsar consume modle 23 

24 .Message Consume->AMQP Basic.consume 24 

25 .Message Consume->AMQP Basic.consume Push实现 – AMQPConsumer <--> Pulsar Consumer • 重写sendMessages()方法 – 消息转换：pulsar message -> AMQP message – 将获取的消息推送到AMQP的客户端 – subscription.addConsumer – consumer.handleFlow 25 

26 .Message Consume->AMQP Basic.get 26 

27 .Message Consume->AMQP Basic.get Pull实现 – Pull 消费流程 • 主动从cursor中读取消息，然后和Push的consumer共享read position • 消息签收的方式和Push方式一致 27 

28 .Message ACK 28 

29 .Message ACK 29