MQTT分享

技术文章

架构

• 整体架构

• mqtt-broker cluster架构

mqtt-locater在内存中存储client到broker的映射。

• mqtt-broker 架构

接口

对外开放接口

• 推送数据

Path: /v1/publish
Verb: POST
Content-Type: application/json
Body: {
    "Topic": "1",
    "Qos": 2,
    "Payload": "cGF5bG9hZA=="
}

• 查询客户端是否在线

Path: /v1/online/:identifier
Verb: GET
Content-Type: application/json
Response: {
  "Online": false
}

二次开发接口

• Authenticate(认证)
• CanPublish(是否有发布权限)
• CanSubscribe(是否有订阅权限)
• OnConnect(上线事件)
• OnDisconnect(下线事件)
• OnSubscribe(订阅事件)
• OnUnsubscribe(取消订阅事件)
• OnPublish(发布事件，只有特定的topic才会通知发布事件到后端的业务系统)

二次开发接口路由配置

[
        {
                "Name": "auth-svr",
                "Balancer": "round-robin",
                "Addrs": [
                        "http://localhost:8080",
                        "http://127.0.0.1:8080"
                ],
                "APIs": [
                        {
                                "Name": "Authenticate",
                                "Method": "",
                                "Path": "/v1/authenticate"
                        },
                        {
                                "Name": "CanPublish",
                                "Method": "",
                                "Path": "/v1/can/publish"
                        },
                        {
                                "Name": "CanSubscribe",
                                "Method": "",
                                "Path": "/v1/can/subscribe"
                        }
                ]
        },
        {
                "Name": "online-svr",
                "Balancer": "sharding",
                "Addrs": [
                        "http://localhost:8080",
                        "http://127.0.0.1:8080"
                ],
                "APIs": [
                        {
                                "Name": "OnConnect",
                                "Method": "POST",
                                "Path": "/v1/on/connect"
                        },
                        {
                                "Name": "OnDisconnect",
                                "Method": "POST",
                                "Path": "/v1/on/disconnect"
                        }
                ]
        },
        {
                "Name": "process-svr",
                "Balancer": "round-robin",
                "Addrs": [
                        "http://localhost:8080",
                        "http://127.0.0.1:8080"
                ],
                "APIs": [
                        {
                                "Name": "OnPublish",
                                "Method": "POST",
                                "Path": "/v1/on/publish"
                        },
                        {
                                "Name": "OnSubscribe",
                                "Method": "POST",
                                "Path": "/v1/on/subscribe"
                        },
                        {
                                "Name": "OnUnsubscribe",
                                "Method": "POST",
                                "Path": "/v1/on/unsubscribe"
                        }
                ]
        }
]

实现

MQTT后台架构设计

协程模型

一个连接两个协程

• 一个协程用于读取、处理MQTT协议
• 一个协程用于Publish消息的下发处理

Publish消息下发

Publish消息的下发主要由以下两个组件来处理：

• 消息队列(缓存要下发的Publish消息)
• 飞行窗口(保存当前正在发送未确认的QoS1/QoS2消息)

队列满时消息的丢弃顺序：先丢弃最早进入队列的Qos0消息，如果没有Qos0消息，则丢弃最早进入的Qos1/Qos2消息。

集群广播

// Publish 发布消息
func (m *Manager) Publish(p *packets.PublishPacket) error {
    req := &proto.Transport{
        Topic:   p.TopicName,
        Qos:     p.FixedHeader.Qos,
        Payload: p.Payload,
    }   
    for _, peer := range m.GetPeers() {
        if peer == m.cluster.self {
            continue
        }   
        c, err := m.pool.GetClient("tcp", peer)
        if err != nil {
            log.Errorw("get client", "error", err)
            continue
        }   
        if err = apis.Transport(c, req); err != nil {
            log.Errorw("transport", "error", err)
            continue
        }   
    }   
    return m.Transport(req)
}

性能

• 单机模式

50W连接，未接入mqtt-backend

40W连接，接入mqtt-backend

• 集群模式

3台机器，105W连接，优化前

2台机器，90W连接，优化后

问题

• 后端接入业务系统后，性能不足
• WR环境的网络不稳定及单机50W连接限制
• 内存问题，50W连接共占用12G内存，平均一个连接占用25K内存，但在开始收发包后，内存上涨到19G

后续

• 将OnConnect等事件通知接口异步化
• 二次开发接口路由表动态化
• 分布式集群模型的改进
• mqtt-direct支持二进制协议
• mqtt-locater支持集群(分片或分布式哈希)
• 日志中支持TraceId,ClientId、配置热加载、更多大盘数据等
• 持续的性能优化

业务系统

IM系统

• 消息发送流程

• 消息接收流程