mqtt简介

规范原则
- 精简
- 发布/订阅模式
- 动态创建主题
- 传输量低，传输率高
- 考虑低带宽、高延迟、不稳定的网络
- 支持连续的会话保持合控制（心跳）
- 客户端计算能力可以低
- 提供服务质量管理
- 不强求传输数据格式类型
服务质量
- QoS0
  - 至多一次
- QoS1
  - 至少一次
- QoS2
  - 只有一次
客户端
- 发布
- 订阅
服务端
- 消息代理Broker
发布订阅
- /
  - 分隔符，分割层级
- - 主题过滤器，通配一个层级
- - 主题过滤器，通配多个层级

mqtt协议

mqtt协议
- 固定报头fixed header
- 可变报文头variable header
- 报文体payload
协议
- 首字节（byte1）
  - bit[4-7]
    - mqtt报文类型
      - connect-0001
      - connack-0010
      - publish-0011
      - puback-0100
      - pubrec-0101
      - …
  - bit[0-3]
    - 标志位(只有少部分类型使用标志位)
    - bit[3]-DUP
      - 1重复
    - bit[2]bit[1]-qos
      - 00-至多一次
      - 01-至少一次
      - 10-只有一次
      - 11-非法消息
    - bit[0]
      - 1保留消息
- 剩余字节(byte2-)

EMQX

mqtt服务器
- curl -s https://assets.emqx.com/scripts/install-emqx-deb.sh | sudo bash
- sudo apt-get install emqx
- sudo emqx start
- Dashboard
  - http://localhost:18083/
  - admin/public

认证

内置数据源
- usrename
- cilent id
外部数据库
- mysql
- ldap
- redis
- mongodb
- postgreSql
其他
- http
- jwt
匿名认证
- low_anonymous = true
  - 默认开启
认证链
- 同时开启多个认证，根据认证插件启动顺序进行链式认证

username认证

管理-插件-emqx_auth_username
emqx/etc/plugins/emqx_auth_username.conf
- auth.user.?.username
- auth.user.?password
api
- get api/v4/auth_username
  - 查询所有用户
- get api/v4/auth_username/${username}
  - 查询具体用户
- post api/v4/auth_username
  - 新增
  - {“username”:“user”,“password”:“123456”}
- put api/v4/auth_username/${username}
  - 更新密码
  - {“password”:“123456”}
- delete api/v4/auth_username/${username}
  - 删除用户
api/v4/auth_username请求头

GET http://hostname:port/api/v4/auth_username HTTP/1.1
Content-Type: application/json
Authorizarion: Basic {{username}}:{{password}}

clientId认证

管理-插件-emqx_auth_clientId
emqx/etc/plugins/emqx_auth_clientid.conf
- auth.client.?.clientid
- auth.client.?.password
api
- get api/v4/auth_clientid
  - 查询所有
- get api/v4/auth_clientid/${clientid}
  - 查询指定
- post api/v4/auth_clientid
  - 新增
  - {“clientid”:“xxx”,“password”:“123456”}
- put api/v4/auth_clientid/${clientid}
  - 更新密码
  - {“password”:“123456”}
- delete api/v4/auth_clientid/${clientid}
  - 删除
api/v4/auth_clientid请求头

POST http://hostname:port/api/v4/auth_clientid HTTP/1.1
Content-Type: application/json
Authorizarion: Basic {{username}}:{{password}}

http认证

管理-插件-emqx_auth_http
emqx/etc/plugins/emqx_auth_http.conf
- auth.http.auth_req
  - 请求认证地址（需要自己实现）
- auth.http.auth_req.method
- auth.http.auth_req.params
- 重试设置
  - auth.http.request.retry_times
  - auth.http.request.retry_interval
  - auth.http.request.retry_backoff

基于springboot的认证服务

AuthController

@RestController
@RequestMapping("/mqtt")
public class AuthController{
    
    @RequestMapping("/auth")
    public ResponseEntity auth(String clientid,String username,String password){
        // 查询数据库
        // 首先通过username查询数据
        String pass = xxxx.xx(username);
        if(StringUtils.isEmpty(pass)){
            return new ResponseEntity(HttpStatus.UNAUTHORIZED);
        }
        if(!pass.equals(password)){
            return new ResponseEntity(HttpStatus.UNAUTHORIZED);
        }
        return new ResponseEntity(HttpStatus.OK);
    }
}

sdk客户端

Paho

依赖

<depenency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
    <version>1.2.2</version>
</depenency>

application.yml

mqtt:
  broker-url: tcp://xxx:xx
  client-id: xx
  username: xx
  password: xx

MqttProperties


@Configuration
@ConfigurationProperties(prefix="mqtt")
@Data
public class MqttProperties{
    private String brokerUrl;
    private String clientId;
    private String username;
    private String password;
}

EmqColent

@Component
public class EmqClient{
    
    private IMqttClient mqttClient;
    @AutoWired
    private MqttProperties mqttProperties;
    @AutoWired
    private MqttCallback mqttCallback;
    
    @PostConstruct
    public void init(){
        MqttClientPersistence mempersitence = new MemoryPersistence();
        try{
            mqttClient = new MqttClient(mqttProperties.getBorkerUrl(),mqttProperties.getClientId,mempersitence);    
        }catch(MqttException ex){
            
        }
    }
    
    // 连接broker
    public void connect(String username,String password){
        MqttConnectOptions options = new MqttConnectOptions();
        options.setAutomaticReconnect(true);
        options.setUserName(username);
        options.setPassword(password);
        options.setCleanSession(true);
        mqttClient.setCallback(mqttCallback);
        try{
            mqttClient.connect(options);
        }catch(MqttException ex){
            
        }
    }
    // 断开连接
    @PreDestroy
    public void disconnect(){
        try{
            mqttClient.disconnect();
        }catch(MqttException ex){
            
        }
    }
    
    // 重连
    public void reConnect(){
        try{
            mqttClient.reconnect();
        }catch(MqttException ex){
            
        }
    }
    
    // 发布
    public void publish(String topic,String msg,QosEnum qos,boolean retain){
        MqttMessage mqttMessage = new MqttMessage();
        mqttMessage.setPayload(msg.getBytes());
        mqttMessage.setQos(qos.value());
        mqttMessage.setRetained(retain);
        try{
            mqttClient.publish(topic,mqttMessage);
        }catch(MqttException ex){
            
        }
    }
    
    // 订阅
    public void subscribe(String topicFilter,QosEnum qos){
      
        try{
            mqttClient.subscribe(topicFilter,qos.value());
        }catch(MqttException ex){
            
        }
    }
    // 取消订阅
    public void unSubscribe(String topicFilter){
        try{
            mqttClient.unsubscribe(topicFilter);
        }catch(MqttException ex){
            
        }
    }
}

QosEnum

public enum QosEnum{
    Qos0(0),Qos1(1),Qos2(2);
    
    private final int value;
    
    QosEnum(int value){
        this.value = value;
    }
    
    public int value(){
        return this.value;
    }
    
}

MessageCallback

public class MessageCallback implements MqttCallback{

    @Override
    // 丢失连接后触发回调
    public void connectionLost(Throwable cause){
        
    }
    
    @Override
    // 应用收到消息后触发回调
    public void messageArrived(String topic,MqttMessage message)throws Exception{
        
    }
    
    @Override
    // 消息发布完成触发回调
    public void deliveryComplete(IMqttDeliveryToken token){
        
    }
}

EmqApplication

@SpringbootApplication
public class EmqApplication{
    public static void main(String[] args){
        SpringApplication.run(EmqApplication.class,agrs);
    }
    
    @Autowired
    private EmqClient emqClient;
    @Autowired
    private MqttProperties properties;
    
    @PostConstruct
    public void init(){
        emqClient.connect(properties.getUsername,properties.getPassword);
        emqClient.subscribe("xxx/#",QosEnum.Qos2);
        // 开启线程,每五秒发布一次
        new Thread(()->{
            while(ture){
                emqClient.public("xxx/123",
                    "public msg :" + LocalDateTime.now().format(DateTimeFormatter.ISO_DATE_TIME),
                    QosEnum.Qos2,
                    false);
                TimeUnit.SECONDS.sleep(5);
            }
        }).start();
        
    }
    
}

mqttjs

https://unpkg.com/mqtt/dist/mqtt.min.js

$(function(){
    // 定义连接
    const options = {
        clean:true,//不保留会话
        connectTimeout:4000,//超时时间
        clientId:'xxx',
        username:'xx',
        password:'xxx'
    }
    
    const connectUrl = 'ws://xxx:8083/mqtt
    const client = mqtt.connect(connectUrl,options);
    
    client.on('reconnect',()=>{
        // 重连
    })
    client.on('close',()=>{
        // 关闭
    })
    client.on('disconnect',(packet)=>{
        // 接收断开连接的数据包
    })
    client.on('offline',()=>{
        // 客户端下线
    })
    client.on('error',(error)=>{
        // 客户端错误
    })
    client.on('packetsend',(packet)=>{
        // 客户端任何消息发出
    })
    client.on('packetreceive',(packet)=>{
        // 客户端任何消息接收
    })
    
    client.on('connect',(connack)=>{
        // 客户端成功连接
        // 订阅
        client.subscribe('xxx/#',{qos:2})
        // 每隔2秒发布一次
        setInterval(publish,2000)
    })
    
    function publish(){
        // 发布数据
        const message = 'h5 message '+ Math.random()+new Date()
        client.publish('xxx/123',message,{qos:2})
    }
    client.on('message',(topic,message,packet)=>{
        // 收到发布的消息事件
    })
    
})

日志输出

emqx.conf
- log.to
  - off
  - file
  - console
  - both(默认)
- log.level
  - warning(默认)
- log.dir
  - log(默认)
- log.file
  - emqx.log(默认)
- log.rotation.size
  - 10MB(默认)
- log.info.flie
  - 将info以上级别写入
- log.error.file
  - 将error日志单独输入
- log.chars_limit
  - 8192
  - 单个日志最大字符
- 日志格式
  - date time level client_info module_info msg
log handler
- 负责日志处理和输出工作进程
- emqx_ctl log handlers list
  - 查询安装的log handler
修改日志级别
- 全局
  - emqx_ctl log set-level debug
- 主日志
  - emqx_ctl log primary-level debug
- 单个log handler
  - emqx_ctl log hanlders set-level file debug
    - id=file的修改为debug

日志追踪

将primary log level 设置为debug
- emqx_ctl log primary-level debug
开启cilentid日志追踪
- emqx_ctl trace start client emq-demo log/emq-demo.log
  - 将clientid=emq-demo的日志输入到emq-demo.log
开启topic日志追踪
- emqx_ctl trace start topic ‘xxx/#’ log/topic_xx.log
  - 将topic=xxx的日志输入到topic_xx.log

ACL

发布订阅权限管理
- 内置ACL
- 数据库ACL
  - MYSQL
  - PostgreSQL
  - Redis
  - MongoDB
- HTTP ACL
超级用户不受acl管理
ACL 规则
- ‘allow/deny’ ‘who’ ‘subscribe/publish’ ‘topics’
etc/emqx.conf
- acl_nomatch(如果acl规则没有匹配上)
  - allow（允许）
  - deny（不允许）
- acl_file(acl文件)
  - etc/acl.conf
- acl_deny_action(acl为禁止的响应)
  - ignore（忽略）/disconnect（断开）
- enable_acl_cache
  - on (开启acl缓存)
- acl_cache_max_size
  = 32(单个客户端最大缓出规则数)
- acl_cache_ttl
  = 1m(默认1分钟，缓存失效)
清除acl缓存
- GET /api/v4/clients/{clientid}/acl_cache
  - 查询缓存
  - Authorization: Basic username:password
- delet /api/v4/clients/{clientid}/acl_cache
  - 删除缓存
  - Authorization: Basic username:password
ACL鉴权链
- 插件开启先后顺序

内置ACL

etc/acl.conf

http ACL

插件
- emqx_auth_http
- etc/plugins/emqx_auth_http.conf
  - 超级用户认证
    - auth.http.super_req
    - auth.http.super_req.method
    - auth.http.super_req.params
  - 其他用户认证
    - auth.http.acl_req
    - auth.http.acl_req.method
    - auth.http.acl_req.params

springboot 代码

@PostMapping("/superuser")
public ResponseEntity superuser(String clientid,String username){
    // 查询当前用户是否时超级用户
    if(clientid.contains("admin")|| username.contains("admin")){
        return new ResponseEntity(HttpStatus.OK);
    }else{
         return new ResponseEntity(HttpStatus.UNAUTHORIZED);
    }
}

@PostMapping("/acl")
// access 1 sub订阅 2 pub发布
public ResponseEntity acl(int access,
    String clientid,
    String username,
    String ipaddr,
    String topic,
    String mountpoint){
    // 查询当前用户是否有topic的操作权限
    if(username.equals("xxx")&& topic.equals("xxx/#")&access == 1){
        return new ResponseEntity(HttpStatus.OK);
    }
    if(clientid.equals("xxx")&& topic.equals("xxx/#")&access == 2){
        return new ResponseEntity(HttpStatus.OK);
    }
    return new ResponseEntity(HttpStatus.UNAUTHORIZED);
}

webhook

插件
- emqx_web_hook
  - 将钩子事件通知到某个web服务
  - /etc/plugins/emqx_web_hook.conf
    - web.hook.api.url
    - web.hook.encode_payload
  - 触发规则
    - web.hook.rule.. =

客户端断开连接事件

// web.hook.rule.client.connected.1={"action":"on_client_connected"}
// web.hook.rule.client.disconnected.1={"action":"on_client_disconnected"}

@RestController
@RequestMapping("/mqtt")
public class WebHookController{
    private Map<String,boolean> clientStatus = new HashMap<>;
    
    @PostMapping("/webhook")
    public void hook(@RequestBody Map<String,Object> params){
        // 获取事件名称
        String action = (String)params.get("action");
        String clientId = (String)params.get("clientid");
        if(action.equals("client_connected")){
            // 客户端接入
            clientStatus.put(clientId,true);
        }
        if(action.equals("client_disconnected")){
            // 客户端下线
             clientStatus.put(clientId,false);
        }
    }
    
    @GetMapping("/status")
    public Map getStatus(){
        return clientStatus;
    }
}

集群

node
- 每一个emxq
- 唯一的节点名称
- etc/emxq.cnf
  - node.name
  - node.cookie
- 订阅表
  - 只存在与订阅者所在节点
  - 主题-订阅者映射
- 路由表
  - 主题-节点映射
- 主题树
  - 根据主题树匹配路由，再根据路由查询路由表
cluster集群
- etc/emqx.cnf
  - cluster.name
  - cluster.proto_dist
  - cluster.discovery
    - 发现策列
集群发现策列
- manual-手动
  - emqx_ctl cluster join xxx
  - emqx_ctl cluster leave
    - 主动退出
  - emqx_ctl cluster force-leave xxx
    - 强制退出
- static-静态
  - cluster.static.seeds
  - 设置集群节点名称列表
- mcast
- dns
- etcd
- k8s

监控api

etc/plugins/emqx_management.conf
- 默认用户
  - management.default_application.id
  - management.default_application.secret
- 管理-应用-新增用户
- 工具-HTTP接口-接口列表

保留消息

每个topic只有一个保留消息
etc/emqx.conf
- mqtt.retain_available=true
  - 开启
etc/plugins/emqx_retainer.conf
- retainer.storage_type
- retainer.max_retained_messages
- retainer.max_payload_size
- retainer.expiry_interval

共享订阅

多个订阅者负载均衡
带群组共享订阅
- $share/
不带群组共享订阅
- 默认都在一个组
- $queue/topic
负载均衡策列
- etc/emqx.conf
  - broker.shared_subscription_strategy
是否需要ack
- 保证通讯质量，一个客户端掉线后发给其他同组客户端
- broker.shared_dispatch_ack_enabled

延迟发布

$delayed/{DelayInterval}/topic
管理-插件-emqx_delayed_publish

代理订阅

客户端自动订阅
etc/emqx.conf
- module.subscription
  - 默认关闭
- 配置规则
  - module.subscription..topic=
  - module.subscription/.qos=
webhook&httpapi实现动态代理订阅
- 客户端上线自动订阅，下线取消订阅

// web.hook.rule.client.connected.1={"action":"on_client_connected"}
// web.hook.rule.client.disconnected.1={"action":"on_client_disconnected"}

@RestController
@RequestMapping("/mqtt")
public class WebHookController{
    private Map<String,boolean> clientStatus = new HashMap<>;
    
    @PostMapping("/webhook")
    public void hook(@RequestBody Map<String,Object> params){
        // 获取事件名称
        String action = (String)params.get("action");
        String clientId = (String)params.get("clientid");
        if(action.equals("client_connected")){
            // 客户端接入
            clientStatus.put(clientId,true);
            autoSub(clientId,"/xx/#",QosEnum.Qos2,true);
        }
        if(action.equals("client_disconnected")){
            // 客户端下线
             clientStatus.put(clientId,false);
             autoSub(clientId,"/xx/#",QosEnum.Qos2,false);
        }
    }
    
    @GetMapping("/status")
    public Map getStatus(){
        return clientStatus;
    }
    
    // 自动订阅和取消
    public void autoSub(String clientId,String topicFilter,QosEnum qos,boolean sub){
        RestTemplate restTemplate = new RestTemplateBuilder().basicAuthentican("admin","public")
            .defaultHeader(HttpHeaders.CONTENT_TYPE,MediaType.APPLICATION_JSON_VALUE)
            .build();
        Map<String,Object> params = new HashMap<>();
        params.put("clientid",clientId);
        params.put("topic",topicFilter);
        params.put("qos",qos.value());
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_JSON);
        HttpEntity entity = new HttpEntity(params,headers);
        if(sub){
            // 订阅
            new Thread(()->{
                ResponseEntit<String> responseEntity = restTemplate.postForEntity("xx/api/v4/mqtt/subscribe",entity,String.class);
            }).start();
            return;
        }
        // 取消
        ResponseEntit<String> responseEntity = restTemplate.postForEntity("xx/api/v4/mqtt/unsubscribe",entity,String.class);

    }
}

主题重写

etc/emqx.conf
- module.rewrite
- 重写规则
  - module.rewrite.rule. = 主题过滤器正则表达式目标表达式
  - 倒叙匹配匹配到第一个就结束

黑名单

少量客户端时使用
etc/emqx.conf
- zone.external.enable_flapping_detect = on
  - 需要开启
- flapping_detect_policy = 离线次数, 检查时间, 封禁时间
  - 短时间多次上下线，客户端就会封禁
api
- get /api/v4/banned
  - 获取黑名单
- post /api/v4/banned
  - {who,as}
  - 添加客户端

速率限制

etc/emqx.conf
- listener.tcp.external.max_conn_rate
- zone.external_publish_limit
- listener.tcp.external.rate_limit

飞行窗口&消息队列

飞行窗口
- 存储已发布未确认消息
消息队列
- 飞行窗口满后存入消息队列
max_infight
max_mqueue_len
mqueue_store_qos0

消息重传

etc/emqx.conf
- retry_interval
  - 消息重传等待间隔

规则引擎

emqx消息与事件等响应规则
- 消息发布
- 事件触发
  - $events/client.connected
应用场景
- 动作监听
- 数据筛选
- 消息路由
- 消息编解码
规则引擎组成
- 规则:{sql语句,动作列表:[{动作,动作参数,绑定资源:{资源配置}}]}
sql语句
- select <字段名> from <主题> [where <条件>]
- foreach <字段名> [do <条件>] [incase <条件>] from <主题> [where <条件>]

系统调优

# 所有进程最大文件数
sysctl -w fs.file-max=2097152
# 单个进程可分割最大文件数
sysctl -w fs.nr_open=2097152
# 允许当前会话/进程打开文件句柄数
ulimit -n 1048576 

# backlog
sysctl -w net.core.somaxconn=32768
sysctl -w net.ipv4.tcp_max_syn_backlog=16384
sysctl -w net.core.netdev_max_backlog=16384

weixin0605

学习笔记-mqtt

mqtt简介

mqtt协议

EMQX

认证

username认证

clientId认证

http认证

基于springboot的认证服务

sdk客户端

Paho

mqttjs

日志输出

日志追踪

ACL

内置ACL

http ACL

springboot 代码

webhook

客户端断开连接事件

集群

监控api

保留消息

共享订阅

延迟发布

代理订阅

主题重写

黑名单

速率限制

飞行窗口&消息队列

消息重传

规则引擎

系统调优