Mochi-MQTT 服务器

English | 简体中文 | 日本語 | 诚招翻译人员！

🎆 mochi-co/mqtt 现已成为新的 mochi-mqtt 组织的一部分。 在此阅读相关公告。

Mochi-MQTT 是一个完全兼容、可嵌入的高性能 Go MQTT v5（和 v3.1.1）代理/服务器

Mochi MQTT 是一个用 Go 编写的可嵌入的完全兼容 MQTT v5 代理服务器，专为开发遥测和物联网项目而设计。该服务器既可以作为独立的二进制文件使用，也可以作为库嵌入到你自己的应用程序中。它的设计目标是尽可能轻量级和快速，同时我们也非常注重确保项目的质量和可维护性。

什么是 MQTT？

MQTT 代表 MQ 遥测传输。它是一种发布/订阅、极其简单和轻量级的消息传输协议，专为受限设备和低带宽、高延迟或不可靠的网络而设计（了解更多）。Mochi MQTT 完全实现了 MQTT 协议的 5.0.0 版本。

Mochi-MQTT 特性

• 完全兼容 MQTTv5 功能，兼容 MQTT v3.1.1 和 v3.0.0：
  • 用户和 MQTTv5 数据包属性
  • 主题别名
  • 共享订阅
  • 订阅选项和订阅标识符
  • 消息过期
  • 客户端会话过期
  • 发送和接收 QoS 流控配额
  • 服务器端断开连接和认证数据包
  • 遗嘱延迟间隔
  • 以及 Mochi MQTT v1 的所有原有 MQTT 功能，如完整的 QoS(0,1,2)、$SYS 主题、保留消息等。
• 面向开发者：
  • 大部分核心代理代码现在都已导出并可访问，实现完全的开发者控制。
  • 全功能且灵活的基于钩子的接口系统，便于"插件"开发。
  • 使用特殊的内联客户端直接注入数据包，或伪装成现有客户端。
• 高性能和稳定：
  • 我们经典的基于 trie 的主题-订阅模型。
  • 客户端特定的写入缓冲区，避免因客户端读取慢或行为不规律而引发的问题。
  • 通过所有 MQTT v5 和 MQTT v3 的 Paho 互操作性测试。
  • 超过一千个经过精心考虑的单元测试场景。
• TCP、Websocket（包括 SSL/TLS）和 $SYS 仪表板监听器。
• 使用钩子内置 Redis、Badger、Pebble 和 Bolt 持久化（你也可以自己开发）。
• 使用钩子内置基于规则的身份验证和 ACL 账本（同样可以自己开发）。

兼容性说明

由于 v5 规范与之前版本的 mqtt 存在重叠，服务器可以同时接受 v5 和 v3 客户端，但请注意，在同时连接 v5 和 v3 客户端的情况下，为 v5 客户端提供的属性和功能将会降级以适应 v3 客户端（例如用户属性）。

对 MQTT v3.0.0 和 v3.1.1 的支持被视为混合兼容。在 v3 规范中没有特别限制的情况下，会使用更现代和安全优先的 v5 行为 - 例如对飞行中和保留消息的过期处理，以及客户端和服务质量流控限制。

这个仓库何时更新？

除非是关键问题，新版本通常会在周末发布。

路线图

• 请提出问题来请求新功能或事件钩子！
• 集群支持。
• 增强的指标支持。

快速开始

使用 Go 运行代理

Mochi MQTT 可以作为独立的代理使用。只需检出此仓库并运行 cmd 文件夹中的 cmd/main.go 入口点，它将暴露 tcp (:1883)、websocket (:1882) 和仪表板 (:8080) 监听器。

cd cmd
go build -o mqtt && ./mqtt

使用 Docker

你现在可以从我们的 Docker 仓库拉取并运行官方 Mochi MQTT 镜像：

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

对于大多数用例，你可以使用基于文件的配置来配置服务器，只需指定一个有效的 yaml 或 json 配置文件。

提供了一个简单的 Dockerfile，用于运行 cmd/main.go 中的 Websocket、TCP 和 Stats 服务器，使用 allow-all 认证钩子。

docker build -t mochi:latest .
docker run -p 1883:1883 -p 1882:1882 -p 8080:8080 -v $(pwd)/config.yaml:/config.yaml mochi:latest

基于文件的配置

你可以在 Docker 镜像中使用基于文件的配置（如上所述），或者通过使用 --config=config.yaml 或 --config=config.json 参数运行构建的二进制文件。

配置文件提供了一种方便的机制，可以轻松准备具有最常见配置的服务器。你可以启用和配置内置钩子和监听器，并指定服务器选项和兼容性：

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

请查看 examples/config 中的示例，了解所有可用的配置选项。

需要注意几个条件：

1. 如果使用基于文件的配置，当前支持配置的钩子类型仅限于 auth、storage 和 debug。每种类型的钩子只能有一个实例。
2. 基于文件的配置只能使用内置钩子，因为服务器需要知道类型和配置结构才能应用它。
3. 只能使用内置监听器，原因同上。

如果需要实现自定义钩子或监听器，请使用 cmd/main.go 中指示的传统方式。

使用 Mochi MQTT 进行开发

作为包导入

将 Mochi MQTT 作为包导入只需要几行代码即可开始使用。

import (
  "log"

  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

  // 清理
}

在 examples 文件夹中可以找到使用各种配置运行代理的示例。

网络监听器

服务器附带了各种预打包的网络监听器，允许代理在不同协议上接受连接。当前的监听器有：

使用 listeners.Listener 接口开发新的监听器。如果你开发了新的监听器，请告诉我们！

可以传递 *listeners.Config 来配置 TLS。 可以在 examples 文件夹或 cmd/main.go 中找到使用示例。

服务器选项和功能

有许多可配置的选项可用于更改行为或限制对服务器某些功能的访问。

server := mqtt.New(&mqtt.Options{
  Capabilities: mqtt.Capabilities{
    MaximumSessionExpiryInterval: 3600,
    MaximumClientWritesPending: 3,
    Compatibilities: mqtt.Compatibilities{
      ObscureNotAuthorized: true,
    },
  },
  ClientNetWriteBufferSize: 4096,
  ClientNetReadBufferSize: 4096,
  SysTopicResendInterval: 10,
  InlineClient: false,
})

查看 mqtt.Options、mqtt.Capabilities 和 mqtt.Compatibilities 结构体以获取完整的选项列表。可以根据需要配置 ClientNetWriteBufferSize 和 ClientNetReadBufferSize 来调整每个客户端的内存使用。Capabilities.MaximumClientWritesPending 的大小将影响服务器的内存使用。如果同时在线的物联网设备数量很大，并且设置的值很大，即使没有数据传输，服务器的内存使用也会大幅增加。默认值为 1024*8，可以根据实际情况调整此参数。

默认配置说明

在决定默认配置时做出了一些需要在此提及的选择：

• 默认情况下，server.Options.Capabilities.MaximumMessageExpiryInterval 的值设置为 86400（24小时），以防止在使用开箱即用的配置时在敌对网络上暴露代理受到DOS攻击（因为无限期的过期时间会允许无限数量的保留/正在传输的消息累积）。如果您在受信任的环境中运行，或者有能力保留更长的时间，您可能希望覆盖此设置（设置为 0 表示永不过期）。

事件钩子

通用事件钩子系统允许开发人员挂接到服务器和客户端生命周期的各个部分，以添加和修改代理的功能。这些通用钩子用于提供从身份验证、持久存储到调试工具的所有功能。

钩子是可堆叠的 - 您可以向服务器添加多个钩子，它们将按添加顺序运行。一些钩子会修改值，这些修改后的值将在返回到运行时代码之前传递给后续的钩子。

许多内部服务器功能现在已经向开发人员公开，因此您可以使用上面的示例创建自己的钩子。如果您这样做了，请提交一个问题让大家知道！

访问控制

允许钩子

默认情况下，Mochi MQTT 使用拒绝所有访问控制规则。要允许连接，必须使用访问控制钩子覆盖此规则。最简单的钩子是 auth.AllowAll 钩子，它为所有连接、订阅和发布提供允许所有规则。它也是最简单的钩子使用方式：

server := mqtt.New(nil)
_ = server.AddHook(new(auth.AllowHook), nil)

如果您将服务器暴露在互联网或不受信任的网络中，请不要这样做 - 它实际上应该仅用于开发、测试和调试。

身份验证账本

身份验证账本钩子提供了一种复杂的机制，用于以结构体格式定义访问规则。身份验证账本规则有两种形式：身份验证规则（连接）和 ACL 规则（发布订阅）。

身份验证规则有 4 个可选标准和一个断言标志：

ACL 规则有 3 个可选标准和一个过滤器匹配：

规则按索引顺序处理（0,1,2,3），在第一个匹配的规则处返回。查看 hooks/auth/ledger.go 以了解结构体。

server := mqtt.New(nil)
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,
        },
      },
    },
  }
})

账本也可以以 JSON 或 YAML 格式存储，并使用 Data 字段加载：

err := server.AddHook(new(auth.Hook), &auth.Options{
    Data: data, // 从字节切片构建账本：yaml 或 json
})

有关更多信息，请参阅 examples/auth/encoded/main.go。

持久存储

Redis

提供了一个基本的 Redis 存储钩子，为代理提供持久性。它可以像任何其他钩子一样添加到服务器，并有几个选项。它在底层使用 github.com/go-redis/redis/v8，可以通过 Options 值完全配置。

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

有关 redis 钩子如何工作或如何使用它的更多信息，请参阅 examples/persistence/redis/main.go 或 hooks/storage/redis 代码。

Pebble DB

如果您更喜欢基于文件的存储，还有一个 Pebble Db 存储钩子。它可以以与其他钩子相似的方式添加和配置（选项较少）。

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

有关 pebble 钩子如何工作或如何使用它的更多信息，请参阅 examples/persistence/pebble/main.go 或 hooks/storage/pebble 代码。

Badger DB

同样，对于基于文件的存储，还有一个 BadgerDB 存储钩子可用。它可以以与其他钩子相似的方式添加和配置。

err := server.AddHook(new(badger.Hook), &badger.Options{
  Path: badgerPath,
})
if err != nil {
  log.Fatal(err)
}

有关 badger 钩子如何工作或如何使用它的更多信息，请参阅 examples/persistence/badger/main.go 或 hooks/storage/badger 代码。

还有一个 BoltDB 钩子已被弃用，改用 Badger，但如果您需要它，请查看 examples/persistence/bolt/main.go。

使用事件钩子进行开发

有许多钩子可用于与代理和客户端生命周期进行交互。 所有钩子和 mqtt.Hook 接口的函数签名可以在 hooks.go 中找到。

最灵活的事件钩子是 OnPacketRead、OnPacketEncode 和 OnPacketSent - 这些钩子可用于控制和修改所有传入和传出的数据包。 | 功能 | 用途 | |------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | OnStarted | 当服务器成功启动时调用。 | | OnStopped | 当服务器成功停止时调用。 | | OnConnectAuthenticate | 当用户尝试与服务器进行身份验证时调用。必须实现此方法以允许或拒绝访问服务器（参见 hooks/auth/allow_all 或 basic）。它可以在自定义钩子中用于检查连接用户是否存在于现有用户数据库中。如果允许则返回 true。 | | OnACLCheck | 当用户尝试发布或订阅主题过滤器时调用。同上。 | | OnSysInfoTick | 当发布 $SYS 主题值时调用。 | | OnConnect | 当新客户端连接时调用，可以返回错误或数据包代码以停止客户端连接过程。 | | OnSessionEstablish | 在新客户端连接并通过身份验证后，以及在建立会话和发送 CONNACK 之前立即调用。 | | OnSessionEstablished | 当新客户端成功建立会话时调用（在 OnConnect 之后）。 | | OnDisconnect | 当客户端因任何原因断开连接时调用。 | | OnAuthPacket | 当收到身份验证数据包时调用。旨在允许开发人员创建自己的 MQTT v5 身份验证数据包处理机制。允许修改数据包。 | | OnPacketRead | 当从客户端收到数据包时调用。允许修改数据包。 | | OnPacketEncode | 在将数据包编码发送给客户端之前立即调用。允许修改数据包。 | | OnPacketSent | 当数据包已发送给客户端时调用。 | | OnPacketProcessed | 当数据包已被接收并成功由代理处理时调用。 | | OnSubscribe | 当客户端订阅一个或多个过滤器时调用。允许修改数据包。 | | OnSubscribed | 当客户端成功订阅一个或多个过滤器时调用。 | | OnSelectSubscribers | 当为主题收集订阅者，但在选择共享订阅订阅者之前调用。允许修改接收者。 | | OnUnsubscribe | 当客户端取消订阅一个或多个过滤器时调用。允许修改数据包。 | | OnUnsubscribed | 当客户端成功取消订阅一个或多个过滤器时调用。 | | OnPublish | 当客户端发布消息时调用。允许修改数据包。 | | OnPublished | 当客户端向订阅者发布消息时调用。 | | OnPublishDropped | 当消息在传递给客户端之前被丢弃时调用，例如客户端响应时间过长。 | | OnRetainMessage | 当发布的消息被保留时调用。 | | OnRetainPublished | 当保留的消息发布给客户端时调用。 | | OnQosPublish | 当向订阅者发出 QoS >= 1 的发布数据包时调用。 | | OnQosComplete | 当消息的 QoS 流程完成时调用。 | | OnQosDropped | 当正在传输的消息在完成前过期时调用。 | | OnPacketIDExhausted | 当客户端用完未使用的数据包 ID 以分配时调用。 | | OnWill | 当客户端断开连接并打算发出遗嘱消息时调用。允许修改数据包。 | | OnWillSent | 当从断开连接的客户端发出 LWT 消息时调用。 | | OnClientExpired | 当客户端会话过期并应被删除时调用。 | | OnRetainedExpired | 当保留的消息过期并应被删除时调用。 | | StoredClients | 返回客户端，例如从持久存储中。 | | StoredSubscriptions | 返回客户端订阅，例如从持久存储中。 | | StoredInflightMessages | 返回正在传输的消息，例如从持久存储中。 | | StoredRetainedMessages | 返回保留的消息，例如从持久存储中。 | | StoredSysInfo | 返回存储的系统信息值，例如从持久存储中。 |

如果您正在构建持久存储钩子，请参考现有的持久钩子以获取灵感和模式。如果您正在构建身份验证钩子，您将需要 OnACLCheck 和 OnConnectAuthenticate。

内联客户端（v2.4.0+）

现在可以通过使用"内联客户端"功能直接从嵌入代码订阅和发布主题。目前，内联客户端不支持共享订阅。内联客户端是作为服务器一部分运行的嵌入式客户端，可以在服务器选项中启用：

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

启用后，您将能够使用 server.Publish、server.Subscribe 和 server.Unsubscribe 方法从代理相邻的代码发出和接收消息。

请参阅 direct examples 以了解实际使用示例。

内联发布

要从嵌入应用程序内部向主题发布基本消息，您可以使用 server.Publish(topic string, payload []byte, retain bool, qos byte) error 方法。

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

在这种情况下，Qos 字节仅用于设置订阅者可用的上限 QoS，符合 MQTT v5 规范。

内联订阅

要从嵌入应用程序内部订阅主题过滤器，您可以使用带有回调函数的 server.Subscribe(filter string, subscriptionId int, handler InlineSubFn) error 方法。请注意，内联订阅仅支持 QoS 0。如果您希望对同一过滤器有多个回调，可以使用 MQTTv5 的 subscriptionId 属性进行区分。

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(filter string, subscriptionId int) error 方法轻松完成此操作：

server.Unsubscribe("direct/#", 1)

数据包注入

如果您想要更多控制，或想设置特定的 MQTT v5 属性和其他值，您可以从您选择的客户端创建自己的发布数据包。这种方法允许您直接将 MQTT 数据包（不仅仅是发布）注入运行时，就像它们是由特定客户端接收的一样。

数据包注入可用于任何 MQTT 数据包，包括 ping 请求、订阅等。由于 Clients 结构和方法现在已导出，您甚至可以代表已连接的客户端注入数据包（如果您有非常特殊的要求）。

大多数情况下，您会希望使用上述描述的内联客户端，因为它具有独特的特权：它绕过所有 ACL 和主题验证检查，这意味着它甚至可以发布到 $SYS 主题。在这种情况下，您可以从头开始创建一个内联客户端，它的行为与内置的内联客户端相同。

cl := server.NewClient(nil, "local", "inline", true)
server.InjectPacket(cl, packets.Packet{
  FixedHeader: packets.FixedHeader{
    Type: packets.Publish,
  },
  TopicName: "direct/publish",
  Payload: []byte("scheduled message"),
})

MQTT 数据包仍需正确构建，因此请参考我们的 测试数据包目录 和 MQTTv5 规范 以获取灵感。

请查看 hooks 示例 以了解此功能的实际应用。

测试

单元测试

Mochi MQTT 通过精心编写的上千个场景的单元测试来确保每个功能都能完全符合我们的预期。您可以使用 go 运行这些测试：

go run --cover ./...

Paho 互操作性测试

您可以使用 examples/paho/main.go 启动代理，然后从 interoperability 文件夹运行 python3 client_test5.py，以检查代理是否符合 Paho 互操作性测试。

请注意，目前 Paho 测试套件中存在一些未解决的假阴性问题，因此在 paho/main.go 示例中启用了某些兼容模式。

性能基准测试

Mochi MQTT 的性能与 Mosquitto、EMQX 等流行代理相当。

性能基准测试使用 MQTT-Stresser 在 Apple Macbook Air M2 上进行，使用 cmd/main.go 的默认设置。考虑到高低吞吐量的突发情况，中位数分数最具参考价值。数值越高越好。

基准测试中呈现的值并不代表真实的每秒消息吞吐量。它们依赖于 mqtt-stresser 的一种不寻常的计算方法，但由于在所有代理中都保持一致，因此仍然可用。 基准测试仅作为一般性能预期指南提供。比较是使用开箱即用的默认配置进行的。 mqtt-stresser -broker tcp://localhost:1883 -num-clients=2 -num-messages=10000 | 代理 | 发布最快 | 中位数 | 最慢 | 接收最快 | 中位数 | 最慢 | | -- | -- | -- | -- | -- | -- | -- | | Mochi v2.2.10 | 124,772 | 125,456 | 124,614 | 314,461 | 313,186 | 311,910 | | Mosquitto v2.0.15 | 155,920 | 155,919 | 155,918 | 185,485 | 185,097 | 184,709 | | EMQX v5.0.11 | 156,945 | 156,257 | 155,568 | 17,918 | 17,783 | 17,649 | | Rumqtt v0.21.0 | 112,208 | 108,480 | 104,753 | 135,784 | 126,446 | 117,108 |

mqtt-stresser -broker tcp://localhost:1883 -num-clients=10 -num-messages=10000

百万消息挑战（立即向服务器发送100万条消息）：

mqtt-stresser -broker tcp://localhost:1883 -num-clients=100 -num-messages=10000

这里EMQX的表现不太确定，可能是docker默认设置不够优化，所以请谨慎看待这个结果，因为我们知道它实际上是一款优秀的软件。

贡献指南

欢迎并鼓励贡献和反馈！打开一个issue来报告bug、提问或提出功能请求。如果你要提交pull request，请尽量遵循以下指南：

• 尽可能保持测试覆盖率。
• 清楚地说明PR的作用和原因。
• 请记得在你做出重要贡献的文件中添加你的SPDX FileContributor标签。

SPDX注释用于以机器可读的格式清楚地标明每个文件的许可证、版权和贡献。如果你要向仓库添加新文件，请确保它有以下SPDX头：

// SPDX-License-Identifier: MIT
// SPDX-FileCopyrightText: 2023 mochi-mqtt
// SPDX-FileContributor: 你的名字或别名 <可选的邮箱地址>

package name

请确保为文件的每个贡献者添加新的SPDX-FileContributor行。参考其他文件的例子。请记得这样做，你对这个项目的贡献是有价值的，我们非常感谢 - 获得认可很重要！

随时间变化的星标数 🥰

你正在项目中使用Mochi MQTT吗？告诉我们！