附上:
喵了个咪的博客: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使用过程中还有很多的细节需要注意,关注细节才能走的更远
注:笔者能力有限有说的不对的地方希望大家能够指出,也希望多多交流!
有疑问加站长微信联系（非本文作者）
