Golang嵌入式高性能MQTT v5/v3代理服务器插件mochi mqtt的使用

Mochi-MQTT是一个完全兼容、可嵌入的高性能Go语言MQTT v5（和v3.1.1）代理服务器/服务器，专为遥测和物联网项目开发而设计。

主要特性

• 完全兼容MQTTv5规范，同时支持MQTT v3.1.1和v3.0.0
• 开发者友好：大部分核心代理代码已导出并可访问
• 高性能且稳定：采用经典的trie主题订阅模型
• 支持TCP、WebSocket（包括SSL/TLS）和$SYS仪表板监听器
• 内置Redis、Badger、Pebble和Bolt持久化存储钩子
• 内置基于规则的认证和ACL账本钩子

快速开始

作为Go包导入使用

import (
  "log"
  "os"
  "os/signal"
  "syscall"

  mqtt "github.com/mochi-mqtt/server/v2"
  "github.com/mochi-mqtt/server/v2/hooks/auth"
  "github.com/mochi-mqtt/server/v2/listeners"
)

func main() {
  // 创建信号通道以便在中断时运行服务器
  sigs := make(chan os.Signal, 1)
  done := make(chan bool, 1)
  signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
  go func() {
    <-sigs
    done <- true
  }()

  // 创建新的MQTT服务器
  server := mqtt.New(nil)
  
  // 允许所有连接
  _ = server.AddHook(new(auth.AllowHook), nil)
  
  // 在标准端口上创建TCP监听器
  tcp := listeners.NewTCP(listeners.Config{ID: "t1", Address: ":1883"})
  err := server.AddListener(tcp)
  if err != nil {
    log.Fatal(err)
  }

  go func() {
    err := server.Serve()
    if err != nil {
      log.Fatal(err)
    }
  }()

  // 运行服务器直到被中断
  <-done

  // 清理
  server.Close()
}

使用Docker运行

docker pull mochimqtt/server
# 或者
docker run -v $(pwd)/config.yaml:/config.yaml mochimqtt/server

文件配置示例

listeners:
  - type: "tcp"
    id: "tcp12"
    address: ":1883"
  - type: "ws"
    id: "ws1"
    address: ":1882"
  - type: "sysinfo"
    id: "stats"
    address: ":1880"
hooks:
  auth:
    allow_all: true
options:
  inline_client: true

事件钩子(Hooks)

Mochi MQTT提供了通用的事件钩子系统，允许开发者钩入服务器和客户端生命周期的各个部分来添加和修改代理功能。

认证钩子示例

err := server.AddHook(new(auth.Hook), &auth.Options{
    Ledger: &auth.Ledger{
    Auth: auth.AuthRules{ // 默认拒绝所有认证
      {Username: "peach", Password: "password1", Allow: true},
      {Username: "melon", Password: "password2", Allow: true},
      {Remote: "127.0.0.1:*", Allow: true},
      {Remote: "localhost:*", Allow: true},
    },
    ACL: auth.ACLRules{ // ACL默认允许所有
      {Remote: "127.0.0.1:*"}, // 本地超级用户允许所有
      {
        // 用户melon可以读写自己的主题
        Username: "melon", Filters: auth.Filters{
          "melon/#":   auth.ReadWrite,
          "updates/#": auth.WriteOnly, // 可以写入updates，但不能读取其他人的updates
        },
      },
      {
        // 否则，没有客户端有发布权限
        Filters: auth.Filters{
          "#":         auth.ReadOnly,
          "updates/#": auth.Deny,
        },
      },
    },
  }
})

持久化存储示例

Redis存储

err := server.AddHook(new(redis.Hook), &redis.Options{
  Options: &rv8.Options{
    Addr:     "localhost:6379", // 默认redis地址
    Password: "",               // 你的密码
    DB:       0,                // 你的redis db
  },
})
if err != nil {
  log.Fatal(err)
}

Pebble DB存储

err := server.AddHook(new(pebble.Hook), &pebble.Options{
  Path: pebblePath,
  Mode: pebble.NoSync,
})
if err != nil {
  log.Fatal(err)
}

内联客户端(Inline Client)

从v2.4.0开始，可以直接从嵌入代码中订阅和发布主题：

server := mqtt.New(&mqtt.Options{
  InlineClient: true,
})

// 发布消息
err := server.Publish("direct/publish", []byte("packet scheduled message"), false, 0)

// 订阅主题
callbackFn := func(cl *mqtt.Client, sub packets.Subscription, pk packets.Packet) {
    server.Log.Info("inline client received message from subscription", "client", cl.ID, "subscriptionId", sub.Identifier, "topic", pk.TopicName, "payload", string(pk.Payload))
}
server.Subscribe("direct/#", 1, callbackFn)

// 取消订阅
server.Unsubscribe("direct/#", 1)

性能基准

Mochi MQTT性能与Mosquitto、EMQX等流行代理相当。以下是基准测试结果示例：

贡献指南

欢迎贡献和反馈！提交PR时请遵循以下准则：

• 尽可能保持测试覆盖率
• 清楚说明PR的作用和原因
• 记得在修改的文件中添加SPDX FileContributor标签

// SPDX-License-Identifier: MIT
// SPDX-FileCopyrightText: 2023 mochi-mqtt
// SPDX-FileContributor: Your name or alias <optional@email.address>

package name

Mochi MQTT是一个功能强大且灵活的MQTT代理服务器，非常适合需要嵌入式MQTT解决方案的Go项目。