EMQX基础功能

# 进入 etc/emqx.conf 
## Value: true | false 
allow_anonymous = true

# 进入etc/plugins/emqx_auth_mysql.conf 

## 不加盐，仅做哈希处理 
auth.mysql.password_hash = sha256 

## salt 前缀：使用 sha256 加密 salt + 密码 拼接的字符串 auth.mysql.password_hash = salt,sha256 

## salt 后缀：使用 sha256 加密 密码 + salt 拼接的字符串 auth.mysql.password_hash = sha256,salt 

## pbkdf2 with macfun iterations dklen 
## macfun: md4, md5, ripemd160, sha, sha224, sha256, sha384, sha512 
## auth.mysql.password_hash = pbkdf2,sha256,1000,20

@hostname = 192.168.200.129 
@port=18083 
@contentType=application/json 
@userName=admin 
@password=public 
#############查看已有用户认证数据############## 
GET http://{{hostname}}:{{port}}/api/v4/auth_username HTTP/1.1 Content-Type: {{contentType}} 
Authorization: Basic {{userName}}:{{password}}

########添加用户认证数据############## 
POST http://{{hostname}}:{{port}}/api/v4/auth_username HTTP/1.1
Content-Type: {{contentType}}
Authorization: Basic {{userName}}:{{password}}

 { "username": "user", "password": "123456" }

###########更改指定用户名的密码############# 
PUT http://{{hostname}}:{{port}}/api/v4/auth_username/user HTTP/1.1 
Content-Type: {{contentType}} 
Authorization: Basic {{userName}}:{{password}} 
 
{ "password": "user" }

###########查看指定用户名信息############# 
GET http://{{hostname}}:{{port}}/api/v4/auth_username/user HTTP/1.1 
Content-Type: {{contentType}} 
Authorization: Basic {{userName}}:{{password}}

###########删除指定的用户信息############# 
DELETE http://{{hostname}}:{{port}}/api/v4/auth_username/user HTTP/1.1 
Content-Type: {{contentType}} 
Authorization: Basic {{userName}}:{{password}}

####添加clientId和密码##### 
POST http://{{hostname}}:{{port}}/api/v4/auth_clientid HTTP/1.1 
Content-Type: {{contentType}} 
Authorization: Basic {{userName}}:{{password}} 

{ "clientid": "emq-client1", "password": "123456" }

#############获取所有详细信息######## 
GET http://{{hostname}}:{{port}}/api/v4/auth_clientid HTTP/1.1 
Content-Type: {{contentType}} 
Authorization: Basic {{userName}}:{{password}}

#############更改指定 Client ID 的密码######## 
PUT http://{{hostname}}:{{port}}/api/v4/auth_clientid/emq-client1 HTTP/1.1 
Content-Type: {{contentType}} 
Authorization: Basic {{userName}}:{{password}} 

{ "password": "654321" }

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.6.5</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.jie</groupId>
    <artifactId>emqx-demo</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>emqx-demo</name>
    <description>Demo project for Spring Boot</description>
    <properties>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-configuration-processor</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <excludes>
                        <exclude>
                            <groupId>org.projectlombok</groupId>
                            <artifactId>lombok</artifactId>
                        </exclude>
                    </excludes>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

server:
  port: 8991
spring:
  application:
    name: emqx-demo

package com.jie.emqxdemo.controller.mqtt;


import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.Map;

/**
 * @description:接受认证请求
 * @author: jie
 * @time: 2022/3/29 20:57
 */
@RestController
@RequestMapping("/mqtt")
@Slf4j
public class AuthController {
    /**
     * 用于存储数据   实际开发应该是存储在数据库中
     */
    private Map<String, String> users;

    @PostConstruct
    public void init() {
        //实际的密码应该是密文,mqtt的http认证组件传输过来的密码是明文，我们需要自己进行加密验证
        users = new HashMap<>();
        users.put("user", "123456");
        users.put("emq-client2", "123456");
        users.put("emq-client3", "123456");
    }

    /**
     * @description:接受消息 如果属性名对应不上可以用@RequestParam注解做映射
     * @author: jie
     * @time: 2022/3/29 21:03
     */
    @PostMapping("/auth")
    public ResponseEntity<?> auth(@RequestParam("clientid") String clientid,
                                  @RequestParam("username") String username,
                                  @RequestParam("password") String password) {
        log.info("emqx认证组件调用自定义的认证服务开始认证,clientid={},username={},password= {}", clientid, username, password);
        //在此处可以进行复杂也的认证逻辑,但是我们为了演示方便做一个固定操作
        String value = users.get(username);
        if (StringUtils.isEmpty(value)) {
            return new ResponseEntity<Object>(HttpStatus.UNAUTHORIZED);
        }
        if (!value.equals(password)) {
            return new ResponseEntity<Object>(HttpStatus.UNAUTHORIZED);
        }
        return new ResponseEntity<Object>(HttpStatus.OK);
    }

}

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

package com.jie.emqxdemo.properties;

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;

/**
 * @description:用于加载yml自定义配置的配置类
 * @author: jie
 * @time: 2022/3/31 21:12
 */
@Configuration
@ConfigurationProperties(prefix = "mqtt")
@Data
public class Mqttproperties {

    private String brokerUrl;

    private String clientId;

    private String username;

    private String password;

    @Override
    public String toString() {
        return "Mqttproperties{" +
                "brokerUrl='" + brokerUrl + '\'' +
                ", clientId='" + clientId + '\'' +
                ", username='" + username + '\'' +
                ", password='" + password + '\'' +
                '}';
    }
}

package com.jie.emqxdemo.client;

import jdk.nashorn.internal.parser.Token;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.stereotype.Component;

/**
 * @description:消息回调
 * @author: jie
 * @time: 2022/3/31 22:12
 */
@Slf4j
@Component
public class MessageCallback implements MqttCallback {


    /**
     * @description:丢失对服务端的连接后触发该方法回调，此处可以做一些特殊处理，比如重连
     * @author: jie 
     * @time: 2022/3/31 22:14
     */
    @Override
    public void connectionLost(Throwable throwable) {
        log.info("丢失了对服务的连接");
    }
    /**
     * @description:应用收到消息后出发的回调
     * 该方法由mqtt客户端同步调用,在此方法未正确返回之前，不会发送ack确认消息到broker
     * 一旦该方法向外抛出了异常客户端将异常关闭，当再次连接时；所有QoS1,QoS2且客户端未进行ack确认的消息都将由 broker服务器再次发送到客户端
     * @author: jie
     * @time: 2022/3/31 22:15
     */
    @Override
    public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
        log.info("订阅到了消息;topic={},messageid={},qos={},msg={}",
                s,
                mqttMessage.getId(),
                mqttMessage.getQos(),
                new String(mqttMessage.getPayload()));
    }
    /**
     * @description:消息发布完成产生的回调
     * @author: jie
     * @time: 2022/3/31 22:18
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        int messageId = iMqttDeliveryToken.getMessageId();
        String[] topics = iMqttDeliveryToken.getTopics();
        log.info("消息发送完成,messageId={},topics={}",messageId,topics);
    }
}

package com.jie.emqxdemo.enums;
/**
 * @description:Qos枚举列
 * @author: jie
 * @time: 2022/3/31 21:54
 */
public enum QosEnum {

    QOS0(0),QOS1(1),QOS2(2);

    private final int value;

    QosEnum(int value){
        this.value = value;
    }
    /**
     * @description:获取枚举值
     * @author: jie
     * @time: 2022/3/31 21:56
     */
    public int value(){
        return this.value;
    }
}

mqtt:
  broker-url: tcp://192.168.58.149:1883
  client-id: emq-client
  username: user
  password: 123456

package com.jie.emqxdemo.client;

import com.jie.emqxdemo.enums.QosEnum;
import com.jie.emqxdemo.properties.Mqttproperties;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.nio.charset.StandardCharsets;

/**
 * @description:客户端封装类
 * @author: jie
 * @time: 2022/3/31 21:02
 */
@Component
@Slf4j
public class EmqClient {

    private IMqttClient iMqttClient;

    /**
     * @description:将配置类注入进来
     */
    @Autowired
    private Mqttproperties mqttproperties;
    @Autowired
    private MqttCallback mqttCallback;

    /**
     * @description:初始化
     * @author: jie
     * @time: 2022/3/31 21:22
     */
    @PostConstruct
    public void init(){
        //持久化机制  持久化到内存  MqttDefaultFilePersistence(持久化到本地的文件系统中)
        MqttClientPersistence mqttClientPersistence = new MemoryPersistence();
        try {
            iMqttClient = new MqttClient(mqttproperties.getBrokerUrl(),mqttproperties.getClientId(),mqttClientPersistence);
        } catch (MqttException e) {
           log.error("初始化客户端mqttClient对象失败,errormsg={} brokerurl={},clientId= {}",e.getMessage(), mqttproperties.getBrokerUrl(),mqttproperties.getClientId());
        }
    }
    /**
     * @description:连接服务端方法
     * @author: jie
     * @time: 2022/3/31 21:33
     */
    public void connect(String username,String password){
        //连接选项对象
        MqttConnectOptions options = new MqttConnectOptions();
        //自动重连
        options.setAutomaticReconnect(true);
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        //临时的会话
        options.setCleanSession(true);
        //方法回调
        iMqttClient.setCallback(mqttCallback);
        try {
            iMqttClient.connect(options);
        } catch (MqttException e) {
           log.error("连接mqtt broker失败,失败原因:{}",e.getMessage());
        }
    }
    /**
     * @description:断开连接
     * @author: jie
     * @time: 2022/3/31 21:48
     */
    @PreDestroy
    public void disConnect(){
        try {
            iMqttClient.disconnect();
        } catch (MqttException e) {
            log.error("断开连接产生的异常,异常信息{}",e.getMessage());
        }
    }
    /**
     * @description:重新连接
     * @author: jie
     * @time: 2022/3/31 21:51
     */
    public void reConnect(){
        try {
            iMqttClient.reconnect();
        } catch (MqttException e) {
            log.error("重新连接失败,失原因:{}",e.getMessage());
        }
    }

    /**
     * @description:发布消息  topic:主题  msg:消息  qos:qos  retain:是否是保留消息
     * @author: jie
      */
    public void publish(String topic, String msg, QosEnum qos,boolean retain){
        MqttMessage mqttMessage = new MqttMessage();
        //设置消息体
        mqttMessage.setPayload(msg.getBytes(StandardCharsets.UTF_8));
        //qos
        mqttMessage.setQos(qos.value());
        try {
            iMqttClient.publish(topic,mqttMessage);
        } catch (MqttException e) {
           log.error("发布消息失败，失败原因:{},topic={},msg={},qos={},retain={}",e.getMessage(),topic,msg,qos,retain);
        }
    }
    /**
     * @description:订阅消息 topicFilter:订阅的主题 qos :qos
     * @author: jie
     * @time: 2022/3/31 22:04
     */
    public void subscribe(String topicFilter,QosEnum qos){
        //订阅主题
        try {
            iMqttClient.subscribe(topicFilter,qos.value());
        } catch (MqttException e) {
            log.error("订阅主题失败，errormsg={},topicFilter:{},qos:{}",e.getMessage(),topicFilter,qos);
        }
    }
    /**
     * @description:取消订阅
     * @author: jie
     * @time: 2022/3/31 22:08
     */
    public void unSubscribe(String topicFilter){
        try {
            iMqttClient.unsubscribe(topicFilter);
        } catch (MqttException e) {
            log.error("取消订阅失败,errormsg={},topicFilter={}",e.getMessage(),topicFilter);
        }
    }

}

package com.jie.emqxdemo;

import com.jie.emqxdemo.client.EmqClient;
import com.jie.emqxdemo.enums.QosEnum;
import com.jie.emqxdemo.properties.Mqttproperties;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import javax.annotation.PostConstruct;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.TimeUnit;

@SpringBootApplication
public class EmqxDemoApplication {

    public static void main(String[] args) {
        SpringApplication.run(EmqxDemoApplication.class, args);
    }
    @Autowired
    private EmqClient emqClient;
    @Autowired
    private Mqttproperties mqttproperties;

    @PostConstruct
    public void init(){
        //连接服务端
         emqClient.connect(mqttproperties.getUsername(),mqttproperties.getPassword());
        //订阅一个主题
        emqClient.subscribe("testtopic/#", QosEnum.QOS2);
        //开启一个新的线程，每隔五秒去向 testtopic/123
        new Thread(()->{
            while (true){
                emqClient.publish("testtopic/123","mqtt msg:"+ LocalDateTime.now().format(DateTimeFormatter.ISO_DATE_TIME),QosEnum.QOS2,false);
                try {
                    TimeUnit.SECONDS.sleep(5);
                }
                catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }
}

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>mqtt.js测试</title>
    <script src="https://cdn.staticfile.org/jquery/1.10.2/jquery.min.js" ></script>
    <script src="https://unpkg.com/mqtt/dist/mqtt.min.js" ></script>
    <style type="text/css">
        div{
            width: 300px;
            height: 300px;
            float: left;
            border: red solid 1px;
        }
    </style>
    <script type="text/javascript">
        $(function (){
            // 连接选项
            const options = { clean: true, // 保留回话
            connectTimeout: 4000, // 超时时间
            // 认证信息
            clientId: 'emqx_client_h5', username: 'user', password: '123456',
            }
            const connectUrl = "ws://192.168.58.149:8083/mqtt";
            const client = mqtt.connect(connectUrl,options);
            /**
             * mqtt.Client相关事件
             */
            //当重新连接启动触发回调
            client.on('reconnect', () => {
                $("#div1").text("正在重连.....");
            });
            //连接断开后触发的回调
            client.on("close",function () {
                $("#div1").text("客户端已断开连接.....");
            });
            //从broker接收到断开连接的数据包后发出。MQTT 5.0特性
            client.on("disconnect",function (packet) {
                $("#div1").text("从broker接收到断开连接的数据包....."+packet);
            });
            //客户端脱机下线触发回调
            client.on("offline",function () {
                $("#div1").text("客户端脱机下线.....");
            });
            //当客户端无法连接或出现错误时触发回调
            client.on("error",(error) =>{
                $("#div1").text("客户端出现错误....."+error);
            });
            //以下两个事件监听粒度细 //当客户端发送任何数据包时发出。这包括published()包以及MQTT用于管理订阅和连接的包
            client.on("packetsend",(packet)=>{
                $("#div1").text("客户端已发出数据包....."+packet);
            });
            //当客户端接收到任何数据包时发出。这包括来自订阅主题的信息包以及MQTT用于管理订阅和连接 的信息包
            client.on("packetreceive",(packet)=>{
                $("#div1").text("客户端已收到数据包....."+packet);
            });
            //成功连接后触发的回调
            client.on("connect",function (connack) {
                $("#div1").text("成功连接上服务器"+new Date());
                //订阅某主题
                /**
                 *  client.subscribe(topic/topic array/topic object, [options], [callback])
                 *  topic:一个string类型的topic或者一个topic数组,也可以是一个对象
                 *  options */
                client.subscribe("testtopic/#",{qos:2});
                //每隔2秒发布一次数据
                setInterval(publish,2000)
            });

            function publish() {
                //发布数据
                /**
                 * client.publish(topic,message,[options], [callback])
                 *  message: Buffer or String
                 *  options:{
                 *  qos:0, //默认0
                 *  retain:false, //默认false
                 *  dup:false, //默认false
                 *  properties:{}
                 *  }
                 *  callback:function (err){}
                 */
                const message = "h5 message "+Math.random()+new Date();
                client.publish("testtopic/123",message,{qos:2});
                $("#div2").text("客户端发布了数据:"+message);
            }
            //当客户端接收到发布消息时触发回调
            /**
             *  topic:收到的数据包的topic
             *  message:收到的数据包的负载playload
             *  packet:收到的数据包 */
            client.on('message', (topic, message,packet) => {
                $("#div3").text("客户端收到订阅消息,topic="+topic+";消息数据:"+message+";数据 包:"+packet);
            });
            //页面离开自动断开连接
            $(window).bind("beforeunload",()=>{
                $("#div1").text("客户端窗口关闭,断开连接");
                client.disconnect();
            })
        })
    </script>
</head>
<body>
    <div id="div1"></div>
    <div id="div2"></div>
    <div id="div3"></div>
</body>
</html>

log.to = both

log.level = warning

log.dir = log

log.file = emqx.log

log.rotation.size = 10MB 
log.rotation.count = 5

log.info.file = info.log

log.error.file = error.log

log.chars_limit = 8192

[Primary Level]		 		-- global log level and filters 
	/ \ 
[Handler 1] [Handler 2]		-- log levels and filters at each handler

emqx_ctl log handlers list 

emqx_ctl log set-level debug

emqx_ctl log primary-level debug

emqx_ctl log handlers set-level file debug

emqx_ctl log primary-level debug 1111

emqx_ctl log primary-level debug debug

emqx_ctl trace start client emq-demo log/emq-demo.log

emqx_ctl log primary-level debug debug

emqx_ctl trace start topic 'testtopic/#' log/topic_testtopic.log trace topic testtopic/# successfully 1234512345