linux服务器centos7搭建mqtt服务java实现mqtt通讯_centos mqtt

安装mqtt服务实现mqtt通讯

---

前言

最近在开发mqtt相关的硬件通讯，自己用虚拟机在linux环境搭了一个mqtt服务器，实现简单的通讯演示

---

一、MQTT通讯简介

1.结构模式图

mqtt通讯有服务端和客户端,服务端统一收集和分发信息。

2.发布和订阅

客户端有发布和订阅两种模式,如mqttClinet1发布的topic=“ASDFG”，mqttClinet2和mqttClient3的订阅topic=“ASDFG”，mqttClinet4的订阅topic=“ASDFGhjkl”，那么当mqttClient1发布一个消息msg=“哈喽!”，那么mqttClient2和mqttClient3的订阅topic和mqttClient1的发布topic一致，将接受到"哈喽!"的消息，mqttClient4就不会接收到任何消息，这就是mqtt的发布和订阅机制。

二、搭建mqtt服务器

1.安装mqtt服务器

可以下载我上传的mqtt安装包:mqtt安装包下载，也可以去官网找适合自己的mqtt安装包:官网emqx多版本mqtt安装包下载，下面我就一我已有的mqtt安装包安装为例进行演示。
进入服务器创建一个emqx文件夹并把emqx.rpm上传到此文件夹下，并安装

[root@localhost ~]# cd /opt/
[root@localhost opt]# mkdir emqx
[root@localhost opt]# cd ./emqx
[root@localhost emqx]# ls
emqx-centos7-4.2.7-x86_64.rpm
[root@localhost emqx]# rpm -ivh emqx-centos7-4.2.7-x86_64.rpm
1
2
3
4
5
6

2.运行mqtt服务器

[root@localhost emqx]#  emqx start
1

上图代表运行成功!

3.开启端口

由于mqtt服务器使用的是1883端口,，同事emqx带可视化界面用到18083端口，确保防火墙将1883和18083这两端口开启,Centos7防火墙端口开启或关闭可以参考我的文章:Centos7防火墙端口开启和关闭

4.访问emqx可视化界面

打开浏览器访问http://主机名:18083

账号是admin，密码是public

进入之后就可以看到注册的mqtt客户端的一些详细信息了

三、测试mqtt发送消息

1.搭建两个mqtt客户端

mqttclient1代码如下

#pom依赖
<dependency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
    <version>1.2.2</version>
</dependency>
1
2
3
4
5
6

public class ClientMQTT {

    public static final String HOST = "tcp://192.168.31.74:1883";
    public static final String TOPIC1 = "ASDFG";//订阅地址
    private static final String clientid = "client1";
    private  MqttClient client;
    private MqttConnectOptions options;

    private String userName = "mqtt1";    //非必须
    private String passWord = "mqtt1";  //非必须

    private void start() {
        try {
            // host为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
            client = new MqttClient(HOST, clientid, new MemoryPersistence());
            // MQTT的连接设置
            options = new MqttConnectOptions();
            // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，设置为true表示每次连接到服务器都以新的身份连接
            options.setCleanSession(false);
            // 设置连接的用户名
            options.setUserName(userName);
            // 设置连接的密码
            options.setPassword(passWord.toCharArray());
            // 设置超时时间 单位为秒
            options.setConnectionTimeout(10);
            // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
            options.setKeepAliveInterval(20);
            //设置断开后重新连接
            options.setAutomaticReconnect(true);
            // 设置回调
            //client.setCallback(new PushCallback());
            MqttTopic topic = client.getTopic(TOPIC1);
            //遗嘱
            options.setWill(topic, "close".getBytes(), 1, true);
            client.connect(options);
            //订阅消息
            int[] Qos = {1};//0：最多一次 、1：最少一次 、2：只有一次
            String[] topic1 = {TOPIC1};
            client.subscribe(topic1, Qos);

            //向client2发送消息
            pubMessage("哈喽!","ASDFG");

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 消息发送
     * @param message
     * @param topic
     */
    public  void pubMessage(String message,String topic){
        MqttMessage mess = new MqttMessage();
        mess.setQos(1);
        mess.setRetained(true);
        mess.setPayload(message.getBytes());
        try {
            client.publish(topic, mess);
        } catch (Exception e) {
            //LOGGER.error(e.getLocalizedMessage());
        }
    }
    
    public static void main(String[] args) {
        ClientMQTT client1 = new ClientMQTT();
        client1.start();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

mqttclient2代码如下

#pom依赖
<dependency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
    <version>1.2.2</version>
</dependency>
1
2
3
4
5
6

public class ClientMQTT {
    public static final String HOST = "tcp://192.168.31.74:1883";
    public static final String TOPIC1 = "ASDFG";//请阅地址
    private static final String clientid = "client2";
    private  MqttClient client;
    private MqttConnectOptions options;

    private String userName = "mqtt2";    //非必须
    private String passWord = "mqtt2";  //非必须

    private void start() {
        try {
            // host为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
            client = new MqttClient(HOST, clientid, new MemoryPersistence());
            // MQTT的连接设置
            options = new MqttConnectOptions();
            // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，设置为true表示每次连接到服务器都以新的身份连接
            options.setCleanSession(false);
            // 设置连接的用户名
            options.setUserName(userName);
            // 设置连接的密码
            options.setPassword(passWord.toCharArray());
            // 设置超时时间 单位为秒
            options.setConnectionTimeout(10);
            // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
            options.setKeepAliveInterval(20);
            //设置断开后重新连接
            options.setAutomaticReconnect(true);
            // 设置回调
            client.setCallback(new PushCallback());
            MqttTopic topic = client.getTopic(TOPIC1);
            //setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息
            //遗嘱
            options.setWill(topic, "close".getBytes(), 1, true);
            client.connect(options);
            //订阅消息
            int[] Qos = {1};//0：最多一次 、1：最少一次 、2：只有一次
            String[] topic1 = {TOPIC1};
            client.subscribe(topic1, Qos);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 消息发送
     * @param message
     * @param topic
     */
    public  void pubMessage(String message,String topic){
        MqttMessage mess = new MqttMessage();
        mess.setQos(1);
        mess.setRetained(true);
        mess.setPayload(message.getBytes());
        try {
            client.publish(topic, mess);
        } catch (Exception e) {
            //LOGGER.error(e.getLocalizedMessage());
        }
    }

    public static void main(String[] args) {
        ClientMQTT client2 = new ClientMQTT();
        client2.start();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

//接收消息回调
class PushCallback implements MqttCallback {

    public void connectionLost(Throwable cause) {

        // 连接丢失后，一般在这里面进行重连
        System.out.println("连接断开，可以做重连");
    }

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

    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()));
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

2.发送消息

首先启动mqttClient2服务准备接收，再启动mqttClient1服务直接发送，下图表示mqttclient2接收到mqttclient1发来的消息