EMQ百万级MQTT消息服务(TLS Docker Golang)

附上:

喵了个咪的博客:w-blog.cn

EMQ官方地址:http://emqtt.com/

EMQ中文文档:http://emqtt.com/docs/v2/guide.html

1.TLS证书验证

为了保障安全.我们常常会使用HTTPS来保障请求不被篡改,作为MQTT使用TLS加密的方式来保障传输安全

EMQ默认使用的TLS加密的端口是8883端口,默认证书在EMQ目录下etc/certs:

对应的配置文件在emq.conf中,可以修改你的端口和配置文件路径

listener.ssl.external = 8883
listener.ssl.external.keyfile = etc/certs/key.pem
listener.ssl.external.certfile = etc/certs/cert.pem

PS:在链接的时候注意需要从之前的链接前缀做如此修改 tcp:// -> ssl://

2.使用Docker部署EMQ

使用Docker部署EMQ很方便版本升级,端口管理和单节点多EMQ等,对于性能来说基本没有带来额外的开销

但是官方并没有提供对于直接可以用的Docker镜像,但是提供的GIT大家可以自己打包以及压缩包,笔者因为使用这里打成了公共进行如下(2.3.5往上版本都会进维护):

registry.cn-hangzhou.aliyuncs.com/sunmi-base/sunmi-emq:2.3.5
registry.cn-hangzhou.aliyuncs.com/sunmi-base/sunmi-emq:2.3.6
registry.cn-hangzhou.aliyuncs.com/sunmi-base/sunmi-emq:2.3.7

当然更重要的一点就是关于配置,EMQ提供通过环境变量的方式影响配置文件,大家可以参考一下daocker-composer和Kubernetes的编排文件,通过环境变量的方式进行配置文件的修改,一下编排配置了mysql鉴权以及自定义鉴权语句和默认开启mysql鉴权插件

version: '2'
services:
 emq:
 image: 'registry.cn-hangzhou.aliyuncs.com/sunmi-base/sunmi-emq:2.3.7'
 ports:
 - '31883:1883'
 - '31083:18083'
 - '38883:8883'
 environment:
 - EMQ_MQTT__ALLOW_ANONYMOUS=false
 - EMQ_AUTH__MYSQL__USERNAME=emq
 - EMQ_AUTH__MYSQL__PASSWORD=Emq666
 - EMQ_AUTH__MYSQL__DATABASE=emq
 - "EMQ_AUTH__MYSQL__SERVER=xxxxxx:3306"
 - "EMQ_AUTH__MYSQL__AUTH_QUERY=select password from mqtt_user where username = '%u' limit 1"
 - "EMQ_AUTH__MYSQL__SUPER_QUERY=select is_superuser from mqtt_user where username = '%u' limit 1"
 - "EMQ_AUTH__MYSQL__ACL_QUERY=select allow, ipaddr, username, clientid, access, REPLACE(topic,'$user','%u') from mqtt_acl where ipaddr = '%a' or username = '%u' or username = '$all' or clientid = '%c'"
 - "EMQ_LOADED_PLUGINS=emq_auth_mysql,emq_recon,emq_modules,emq_retainer,emq_dashboard"
 restart: always

以下是K8S编排文件:

apiVersion: extensions/v1beta1 # K8S对应的API版本
kind: Deployment # 对应的类型
metadata:
 name: emq-deployment
 labels:
 name: emq-deployment
spec:
 replicas: 1 # 镜像副本数量
 template:
 metadata:
 labels: # 容器的标签 可和service关联
 app: emq
 spec:
 containers:
 - name: emq # 容器名和镜像
 image: registry.cn-hangzhou.aliyuncs.com/sunmi-base/sunmi-emq:2.3.6
 imagePullPolicy: Always
 env: # 环境变量
 - name: EMQ_MQTT__ALLOW_ANONYMOUS
 value: "false"
 - name: EMQ_AUTH__MYSQL__USERNAME
 value: "emq"
 - name: EMQ_AUTH__MYSQL__PASSWORD
 value: "Emq666"
 - name: EMQ_AUTH__MYSQL__DATABASE
 value: "emq"
 - name: EMQ_AUTH__MYSQL__SERVER
 value: "xxxxxx:3306"
 - name: EMQ_AUTH__MYSQL__AUTH_QUERY
 value: "select password from mqtt_user where username = '%u' limit 1"
 - name: EMQ_AUTH__MYSQL__SUPER_QUERY
 value: "select is_superuser from mqtt_user where username = '%u' limit 1"
 - name: EMQ_AUTH__MYSQL__ACL_QUERY
 value: "select allow, ipaddr, username, clientid, access, REPLACE(topic,'$user','%u') from mqtt_acl where ipaddr = '%a' or username = '%u' or username = '$all' or clientid = '%c'"
 - name: EMQ_LOADED_PLUGINS
 value: "emq_auth_mysql,emq_recon,emq_modules,emq_retainer,emq_dashboard"
---
apiVersion: v1
kind: Service
metadata:
 name: emq-service # 名称
 labels:
 name: emq-service
spec:
 type: NodePort # 开发端口的类型
 selector: # service负载的容器需要有同样的labels
 app: emq
 ports:
 - name: emq-service-1883-30111
 port: 1883 # 通过service来访问的端口
 targetPort: 1883 # 对应容器的端口
 nodePort: 30111
 - name: emq-service-8883-30112
 port: 8883 # 通过service来访问的端口
 targetPort: 8883 # 对应容器的端口
 nodePort: 30112
 - name: emq-service-18083-30113
 port: 18083 # 通过service来访问的端口
 targetPort: 18083 # 对应容器的端口
 nodePort: 30113

PS:需要在宿主机做好TCP链路优化

3.Goalng客户端

笔者这边使用的是gobot库基于https://github.com/eclipse/paho.mqtt.golang paho体系下的库,例子如下:

package main
import (
 "gobot.io/x/gobot"
 "gobot.io/x/gobot/platforms/mqtt"
 "fmt"
 "time"
)
func main() {
 mqttAdaptor := mqtt.NewAdaptor("tcp://0.0.0.0:1883", "pinger")
 work := func() {
 mqttAdaptor.On("hello", func(msg mqtt.Message) {
 fmt.Println(msg)
 })
 mqttAdaptor.On("hola", func(msg mqtt.Message) {
 fmt.Println(msg)
 })
 data := []byte("o")
 gobot.Every(1*time.Second, func() {
 mqttAdaptor.Publish("hello", data)
 })
 gobot.Every(5*time.Second, func() {
 mqttAdaptor.Publish("hola", data)
 })
 }
 robot := gobot.NewRobot("mqttBot",
 []gobot.Connection{mqttAdaptor},
 work,
 )
 robot.Start()
}

使用 mqttAdaptor.Publish可以发送消息 mqttAdaptor.On 可以订阅消息,如果有用户验证可以使用如下方式:

	mqttAdaptor = mqtt.NewAdaptorWithAuth(
		"EMQ.host",
		"EMQ.clientID",
		"EMQ.userName",
		"EMQ.passWordActive",
	)

断开也可以配置重连规则自动重连(默认未开启,作为服务端强烈建议开启)

mqttAdaptor.SetAutoReconnect(true)

以及消息清理机制(默认断开连接清理消息)

mqttAdaptor.SetCleanSession(false)

也可以指定使用TLS证书连接

mqttAdaptor.SetUseSSL(true)
# 下面可以指定证书(如果EMQ使用了标准的CA证书下面就不用配置了)
mqttAdaptor.SetClientKey(`/client/client-key.pem`)
mqttAdaptor.SetClientCert(`/client/client-cert.pem`)

4. 总结

在EMQ和MQTT使用过程中还有很多的细节需要注意,关注细节才能走的更远

注:笔者能力有限有说的不对的地方希望大家能够指出,也希望多多交流!

本文来自:开源中国博客

感谢作者:喵了_个咪

查看原文:EMQ百万级MQTT消息服务(TLS Docker Golang)

• 编辑
• 预览