Emitter-io: 高性能分布式MQTT/Websocket发布订阅平台

Emitter-io是一个高性能的分布式MQTT和Websocket发布/订阅平台，专为实时通信设计。它提供了低延迟、高吞吐量的消息传递能力，非常适合IoT、实时应用和微服务通信场景。

主要特性

• 支持MQTT协议和Websocket协议
• 分布式架构，支持水平扩展
• 持久化消息存储
• 基于频道的安全模型
• 低延迟、高吞吐量

安装Emitter

go get github.com/emitter-io/go

基本使用示例

1. 连接到Emitter服务器

package main

import (
	"fmt"
	"log"
	"time"

	"github.com/emitter-io/go"
)

func main() {
	// 创建Emitter客户端
	client, err := emitter.Connect("tcp://127.0.0.1:8080", func(_ *emitter.Client, msg emitter.Message) {
		fmt.Printf("收到消息: %s\n", msg.Payload())
	})
	
	if err != nil {
		log.Fatal(err)
	}
	defer client.Disconnect(0)
	
	// 其他操作...
}

2. 发布消息

// 发布消息到指定频道
key := "your-channel-key" // 需要替换为实际的key
channel := "demo-channel"
message := "Hello, Emitter!"

// 发布消息
err = client.Publish(key, channel, message)
if err != nil {
	log.Fatal(err)
}

3. 订阅频道

// 订阅频道
_, err = client.Subscribe(key, channel, func(client *emitter.Client, msg emitter.Message) {
	fmt.Printf("收到订阅消息: %s\n", msg.Payload())
})
if err != nil {
	log.Fatal(err)
}

// 保持连接
time.Sleep(30 * time.Second)

高级功能

1. 使用通配符订阅

// 订阅所有以demo开头的频道
_, err = client.Subscribe(key, "demo/+", func(client *emitter.Client, msg emitter.Message) {
	fmt.Printf("通配符订阅收到消息: %s from %s\n", msg.Payload(), msg.Channel())
})

2. 消息持久化

// 发布持久化消息 (在频道名后加"/"和保留时间)
err = client.Publish(key, "demo-channel/10m", "This message will persist for 10 minutes")

3. 安全控制

Emitter使用基于密钥的访问控制：

// 生成只读密钥
readKey, err := client.GenerateKey("master-key", "demo-channel", emitter.AllowRead)
if err != nil {
	log.Fatal(err)
}
fmt.Println("Read-only key:", readKey)

// 生成读写密钥
writeKey, err := client.GenerateKey("master-key", "demo-channel", emitter.AllowRead|emitter.AllowWrite)
if err != nil {
	log.Fatal(err)
}
fmt.Println("Read-write key:", writeKey)

性能优化建议

1. 连接池: 对于高并发场景，使用连接池管理Emitter连接
2. 批量发布: 合并小消息为批量消息减少网络开销
3. QoS选择: 根据场景选择合适的QoS级别
4. 合理使用通配符: 避免过于宽泛的通配符订阅

完整示例

package main

import (
	"fmt"
	"log"
	"time"

	"github.com/emitter-io/go"
)

func main() {
	// 1. 连接到Emitter服务器
	client, err := emitter.Connect("tcp://127.0.0.1:8080", nil)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Disconnect(0)

	// 2. 生成密钥
	const masterKey = "your-master-key"
	const channel = "demo-channel"
	
	// 生成订阅密钥
	subKey, err := client.GenerateKey(masterKey, channel, emitter.AllowRead)
	if err != nil {
		log.Fatal(err)
	}
	
	// 生成发布密钥
	pubKey, err := client.GenerateKey(masterKey, channel, emitter.AllowWrite)
	if err != nil {
		log.Fatal(err)
	}

	// 3. 订阅频道
	_, err = client.Subscribe(subKey, channel, func(_ *emitter.Client, msg emitter.Message) {
		fmt.Printf("收到消息: %s\n", msg.Payload())
	})
	if err != nil {
		log.Fatal(err)
	}

	// 4. 发布消息
	for i := 0; i < 5; i++ {
		msg := fmt.Sprintf("消息 %d", i+1)
		err = client.Publish(pubKey, channel, msg)
		if err != nil {
			log.Fatal(err)
		}
		time.Sleep(1 * time.Second)
	}
}

部署建议

1. 集群部署: 对于生产环境，建议部署Emitter集群
2. 监控: 监控消息吞吐量和延迟
3. 持久化配置: 根据业务需求配置适当的消息保留策略

Emitter-io是一个功能强大且灵活的发布/订阅平台，通过合理使用可以构建高性能的实时通信系统。