Windows上安装EMQX作为MQTT服务器以及配置MQTTX客户端

一、安装EMQX MQTT服务器

EMQX 是一个高性能、可扩展的物联网消息中间件，EMQX 实现了 MQTT 协议的服务器端，即 MQTT Broker，它负责接收来自客户端的连接请求，处理订阅、发布消息，并将消息转发给相应的订阅者。

下载EMQX

EMQX官方网站 https://www.emqx.com/zh/try?product=broker 提供了EMQX开源版软件包

但是5.3.2版本以后就没有提供Windows系统软件包，需要到这里下载https://www.emqx.com/zh/downloads/broker

在这里插入图片描述
解压

在这里插入图片描述

配置EMQX，EMQX的配置文件位于etc\emqx.conf，可以根据需要修改配置文件，例如更改端口号、添加插件等

在这里插入图片描述

启动EMQX，进入\bin目录下，启动cmd输入emqx start

在这里插入图片描述
检查启动状态emqx_ctl status

可以在任务管理器中看到erl.exe是否运行

在这里插入图片描述

访问EMQX Dashboard，启动EMQX 后，进入http://127.0.0.1:18083中，帐号admin，密码public
8083 MQTT/WebSocket 端口
18083 EMQX Dashboard 管理控制台端口
在这里插入图片描述

登录后右上角设置，可以切换成简体中文

在这里插入图片描述

右侧栏第一个菜单，可查看集群概览

在这里插入图片描述

右侧栏第二个菜单，选择客户端认证，创建客户端认证，选择内置数据库，默认参数配置

在这里插入图片描述

选择用户管理，添加用户账号：admin 密码：public

在这里插入图片描述

二、安装及配置MQTTX客户端

MQTTX 则是一个轻量级的 MQTT 客户端工具，主要用于测试和调试 MQTT 应用程序。

下载MQTTX

访问MQTTX的官方下载页面https://mqttx.app/zh

在这里插入图片描述

双击下载的exe文件，选择安装目录并点击完成

在这里插入图片描述

配置连接到EMQX

在MQTTX中点击“+”添加新的连接，输入连接名称，如“EMQX Server”，服务器地址填写为 localhost 或你的EMQX服务器的实际IP地址，端口默认为 1883，如果EMQX配置了SSL/TLS，应使用 8883，输入用户名和密码（如果EMQX配置了认证）

在这里插入图片描述

点击连接，可以开始订阅主题、发布消息等操作来测试MQTT通信。
在这里插入图片描述

在这里插入图片描述
在EMQX Dashboard可以看到客户端已经连接成功。

进入到WebSocket 客户端，可以进行订阅和发布的测试
在这里插入图片描述

三、EMQX目录结构

引用自：https://www.cnblogs.com/yking/p/15702684.html

不同安装方式得到的 EMQ X 其目录结构会有所不同，具体如下：

描述	使用 ZIP 压缩包安装	使用二进制包安装	Homebrew(MacOS)安装
可执行文件目录	`./bin`	`/usr/lib/emqx/bin`	`/usr/local/bin`
数据文件	`./data`	`/var/lib/emqx/data`	`/usr/local/Cellar/emqx/*/data`
Erlang 虚拟机文件	`./erts-*`	`/usr/lib/emqx/erts-*`	`/usr/local/Cellar/emqx/*/erts-`
配置文件目录	`./etc`	`/etc/emqx/etc`	`/usr/local/Cellar/emqx/*/etc`
依赖项目录	`./lib`	`/usr/lib/emqx/lib`	`/usr/local/Cellar/emqx/*/lib`
日志文件	`./log`	`/var/log/emqx`	`/usr/local/Cellar/emqx/*/log`
启动相关的脚本、schema 文件	`./releases`	`/usr/lib/emqx/releases`	`/usr/local/Cellar/emqx/*/releases`

以上目录中，用户经常接触与使用的是 bin、etc、data、log 目录。

bin 目录

emqx、emqx.cmd

EMQX 的可执行文件，具体使用可以查看基本命令。

emqx_ctl、emqx_ctl.cmd

EMQX 管理命令的可执行文件，具体使用可以查看管理命令 CLI。

etc 目录

EMQX 通过 etc 目录下配置文件进行设置，主要配置文件包括:

配置文件	说明
emqx.conf	EMQX 配置文件
acl.conf	EMQX 默认 ACL 规则配置文件
plugins/*.conf	EMQX 各类插件配置文件
certs	EMQX SSL 证书文件
emqx.lic	License 文件仅限 EMQX Enterprise

EMQX 具体的配置内容可以查看配置项。

data 目录

EMQX 将运行数据存储在 data 目录下，主要的文件包括:

configs/app.*.config

EMQX 读取 etc/emqx.conf 和 etc/plugins/*.conf 中的配置后，转换为 Erlang 原生配置文件格式，并在运行时读取其中的配置。

loaded_plugins

loaded_plugins 文件记录了 EMQ X 默认启动的插件列表，可以修改此文件以增删默认启动的插件。loaded_plugins 中启动项格式为 {<Plugin Name>, <Enabled>}.，<Enabled> 字段为布尔类型，EMQX 会在启动时根据 <Enabled> 的值判断是否需要启动该插件。关于插件的更多内容，请查看插件。

$ cat loaded_plugins
{emqx_management,true}.
{emqx_recon,true}.
{emqx_retainer,true}.
{emqx_dashboard,true}.
{emqx_rule_engine,true}.
{emqx_bridge_mqtt,false}.

Mnesia

Mnesia 数据库是 Erlang 内置的一个分布式 DBMS，可以直接存储 Erlang 的各种数据结构。

EMQX 使用 Mnesia 数据库存储自身运行数据，例如告警记录、规则引擎已创建的资源和规则、Dashbaord 用户信息等数据，这些数据都将被存储在 mnesia 目录下，因此一旦删除该目录，将导致 EMQX 丢失所有业务数据。

可以通过 emqx_ctl mnesia 命令查询 EMQX 中 Mnesia 数据库的系统信息，具体请查看管理命令 CLI。

log 目录

emqx.log.*

EMQX 运行时产生的日志文件，具体请查看日志与追踪。

crash.dump

EMQX 的崩溃转储文件，可以通过 etc/emqx.conf 修改配置，具体内容可以查看配置项。

erlang.log.*

以 emqx start 方式后台启动 EMQX 时，控制台日志的副本文件。

四、使用 Java SDK 连接

引用自：https://docs.emqx.com/zh/emqx/latest/connect-emqx/java.html

Eclipse Paho Java Client 是用 Java 编写的 MQTT 客户端库（MQTT Java Client），可用于 JVM 或其他 Java 兼容平台（例如Android）。

Eclipse Paho Java Client 提供了MqttAsyncClient 和 MqttClient 异步和同步 API。

通过 Maven 安装 Paho Java

通过包管理工具 Maven 可以方便地安装 Paho Java 客户端库，截止目前最新版本安装如下：

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

Paho Java 使用示例

Java 体系中 Paho Java 是比较稳定、广泛应用的 MQTT 客户端库，本示例包含 Java 语言的 Paho Java 连接 EMQX Broker，并进行消息收发完整代码：

App.java

package io.emqx;

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;


public class App {
    public static void main(String[] args) {
        String subTopic = "testtopic/#";
        String pubTopic = "testtopic/1";
        String content = "Hello World";
        int qos = 2;
        String broker = "tcp://broker.emqx.io:1883";
        String clientId = "emqx_test";
        MemoryPersistence persistence = new MemoryPersistence();

        try {
            MqttClient client = new MqttClient(broker, clientId, persistence);

            // MQTT 连接选项
            MqttConnectOptions connOpts = new MqttConnectOptions();
            connOpts.setUserName("emqx_test");
            connOpts.setPassword("emqx_test_password".toCharArray());
            // 保留会话
            connOpts.setCleanSession(true);

            // 设置回调
            client.setCallback(new PushCallback());

            // 建立连接
            System.out.println("Connecting to broker: " + broker);
            client.connect(connOpts);

            System.out.println("Connected");
            System.out.println("Publishing message: " + content);

            // 订阅
            client.subscribe(subTopic);

            // 消息发布所需参数
            MqttMessage message = new MqttMessage(content.getBytes());
            message.setQos(qos);
            client.publish(pubTopic, message);
            System.out.println("Message published");

            client.disconnect();
            System.out.println("Disconnected");
            client.close();
            System.exit(0);
        } catch (MqttException me) {
            System.out.println("reason " + me.getReasonCode());
            System.out.println("msg " + me.getMessage());
            System.out.println("loc " + me.getLocalizedMessage());
            System.out.println("cause " + me.getCause());
            System.out.println("excep " + me);
            me.printStackTrace();
        }
    }
}

回调消息处理类 OnMessageCallback.java

package io.emqx;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;

public class OnMessageCallback implements MqttCallback {
    public void connectionLost(Throwable cause) {
        // 连接丢失后，一般在这里面进行重连
        System.out.println("连接断开，可以做重连");
    }

    public void messageArrived(String topic, MqttMessage message) throws Exception {
        // subscribe后得到的消息会执行到这里面
        System.out.println("接收消息主题:" + topic);
        System.out.println("接收消息Qos:" + message.getQos());
        System.out.println("接收消息内容:" + new String(message.getPayload()));
    }

    public void deliveryComplete(IMqttDeliveryToken token) {
        System.out.println("deliveryComplete---------" + token.isComplete());
    }
}