eemqtt

README

eemqtt 组件使用指南

Table of contents

• 基本组件
  • 快速上手
• 使用说明
  • 注意事项
• 测试

基本组件

对 eclipse/paho.golang 进行了轻量封装，并提供了以下功能：

• 规范了标准配置格式，提供了统一的 Load().Build() 方法。
• 开启 Debug 后可输出 调试信息至终端。
• 提供了监控拦截器，连接失败计数。
• paho.golang 包 默认支持 MQTT 3.11/5.0 协议，支持断开重连。

快速上手

使用样例可参考 example

注意事项

官网指定 mqtt SDKpaho.mqtt.golang 本组件是另一个分支，支持 5.0协议，但并没有完全支持，作者准备自己完善

使用说明

通过连接到 emqtt服务端 实现发送主题消息 / 订阅主题消息

配置文件说明

[emqtt]
debug = true #启用框架自动调试日志
brokers = ["tcp://127.0.0.1:1883"]
#关闭匿名认证后，需要设置用户名和密码才能连接 EMQX_ALLOW_ANONYMOUS=false 
username = ""
password = ""

#通过 tls 安全认证 true 为启用
EnableTLS = false

#如果未设置 客户端正式 只需要配置 ca 文件即可
TLSClientCA = "./certs/cacert.pem"

#订阅的主题配置， 本次配置了两个主题，连接到服务器后将自定订阅这两个主题
[emqtt.subscribeTopics.s1]
topic = "topic1"
qos = 0

[emqtt.subscribeTopics.s2]
topic = "topic2"
qos = 0

代码说明

  ...
    //统一的初始化组件方式
	emqClient = eemqtt.Load("emqtt").Build()
    
    //start 组件自动连接服务器，并订阅配置主题     
	emqClient.Start(msgHandler)
     
	//给主题 topic1 发送消息
    emqClient.PublishMsg("topic1", 0, msg)


	//msgHandler 收到订阅主题的消息处理函数
    func msgHandler(ctx context.Context, pp *paho.Publish) {
       elog.Info("receive meg", elog.Any("topic", pp.Topic), elog.Any("msg", string(pp.Payload)))
       //todo 做相关的业务处理
    }

  ...

注意事项

发送给主题的消息，一定是string 或 []byte 类型
对象需要转换一下

//转换后发送 
bytes, _ := json.Marshal(message) 
emqClient.PublishMsg("topic1", 0, bytes) 

关于emqtt 相关的概念 例如qos..等等不了解的请自行去 emqx官网 学习

测试

在线测试服务

emqx 官网提供的测试服务免费的在线 MQTT 5 服务器
请参考案例 example

配置信息

[emqtt]
debug = true #启用框架自动调试日志

#通过官网服务进行测试 https://www.emqx.com/zh/mqtt/public-mqtt5-broker
brokers = ["mqtts://broker-cn.emqx.io:8883"]

#启用证书
EnableTLS = true
TLSClientCA = "./certs/broker.emqx.io-ca.crt"



[emqtt.subscribeTopics.s1]
topic = "topic1"
qos = 1

[emqtt.subscribeTopics.s2]
topic = "topic2"
qos = 1

自己搭建测试服务

为了方便大家测试，这里提供了一个搭建容器测试环境的案例

1. 安装redis

requirepass 设置 redis 初始化访问密码 案例中设置为 root

docker run -d --name redis -p 6379:6379 redis:latest redis-server --requirepass "root"

2. 安装 emqx 服务： 默认启用密码验证 redis 插件

EMQX_AUTH__REDIS__SERVER 是 redis的服务器和端口，填写宿主机真实IP
EMQX_AUTH__REDIS__PASSWORD redis 访问密码
EMQX_AUTH__REDIS__PASSWORD_HASH=plain 为了方便测试 密码配置成明文形式
EMQX_ALLOW_ANONYMOUS = false 默认关闭了 匿名认真，必须输入 用户名和密码才能登录

docker run -d --name emqx -p 18083:18083 -p 1883:1883 -p 4369:4369 -p 8083:8083 -p 8084:8084  \
    -e EMQX_LISTENER__TCP__EXTERNAL=1883 \
    -e EMQX_LOADED_PLUGINS="emqx_auth_redis,emqx_recon,emqx_retainer,emqx_management,emqx_dashboard" \
    -e EMQX_AUTH__REDIS__SERVER="192.168.1.100:6379" \
    -e EMQX_AUTH__REDIS__PASSWORD="root" \
    -e EMQX_AUTH__REDIS__PASSWORD_HASH=plain \
    -e EMQX_ALLOW_ANONYMOUS="false" \
    emqx/emqx:v4.0.0

3. 设置emqx 访问密码

连接上redis命令行，分别执行下面的两条语句设置用户名和密码。
用户名换成你的用户名，密码换成你的密码。

  HSET mqtt_user:用户名 is_superuser 1
  HSET mqtt_user:用户名 password 密码

4.go do it

Documentation

Index

• Constants
• func DefaultConfig() *config
• type Component
  • func (c *Component) Client() *autopaho.ConnectionManager
  • func (c *Component) PublishMsg(topic string, qos byte, payload interface{})
  • func (c *Component) Start(handler OnPublishHandler)
  • func (c *Component) Stop()
• type Container
  • func DefaultContainer() *Container
  • func Load(key string) *Container
  • func (c *Container) Build(options ...Option) *Component
• type Message
• type OnPublishHandler
• type Option
  • func WithClientID(clientID string) Option
  • func WithPassword(password string) Option
  • func WithTLSSessionCache(tsc tls.ClientSessionCache) Option
  • func WithUsername(username string) Option

Constants

const PackageName = "component.eemqtt"

Functions

func DefaultConfig

func DefaultConfig() *config

DefaultConfig 返回默认配置

Types

type Component

type Component struct {
	ServerCtx context.Context

	Brokers string
	// contains filtered or unexported fields
}

Component ...

func (*Component) Client

func (c *Component) Client() *autopaho.ConnectionManager

func (*Component) PublishMsg

func (c *Component) PublishMsg(topic string, qos byte, payload interface{})

func (*Component) Start

func (c *Component) Start(handler OnPublishHandler)

Start 开始启用

func (*Component) Stop

func (c *Component) Stop()

type Container

type Container struct {
	// contains filtered or unexported fields
}

func DefaultContainer

func DefaultContainer() *Container

func Load

func Load(key string) *Container

func (*Container) Build

func (c *Container) Build(options ...Option) *Component

Build 构建组件

type Message

type Message struct {
	Count uint64
}

type OnPublishHandler

type OnPublishHandler = func(ctx context.Context, pp *paho.Publish)

订推消息处理

type Option

type Option func(c *Container)

func WithClientID

func WithClientID(clientID string) Option

func WithPassword

func WithPassword(password string) Option

func WithTLSSessionCache

func WithTLSSessionCache(tsc tls.ClientSessionCache) Option

func WithUsername

func WithUsername(username string) Option