MQTT：java实现：

一、首先搭建MQTT服务器

1、服务器下载地址：代理服务器
http://archive.apache.org/dist/activemq/activemq-apollo/1.7.1/
安装过程：
2、解压后的bin文件夹下：cmd 执行命令：
apollo.cmd create mybroker
然后会自动创建一个mybroker的文件夹：为刚刚命令后取的名字
3、再进入mybroke文件夹里面，再执行cmd 命令：
apollo-broker.cmd run
然后访问网页：http://127.0.0.1:61680/console/index.html#
默认账户：admin
默认密码：password
代理服务器就启动成功了！

二、实现java代码：

其实服务端和客户端都是mqtt服务器的客户，只是一个发布，一个接收：
依赖：

<?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.4.5</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.example</groupId>
    <artifactId>dtest</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>dtest</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-starter-test</artifactId>
            <scope>test</scope>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>

        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-mqtt</artifactId>
            <version>5.4.6</version>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

服务端代码：

package com.example.dtest.mqtt;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
public class MqttService {
    //tcp: MQTT安装的服务器地址 ：MQTT定义的端口
    public static final String HOST = "tcp://0.0.0.0:61613";
    //定义一个主题
    public static final String TOPIC = "topic11";
    //定义MQTT的ID，可以在MQTT服务配置中指定
    public static final String clentid = "server11";

    private MqttClient client;
    private MqttTopic topic11;
    private String userName = "admin";
    private String passWord = "password";

    private MqttMessage message;

    /**
     * 构造函数
     * @throws MqttService
     */
    public MqttService() throws MqttException {

        // MemoryPersistence设置clientid的保存形式，默认为以内存保存
        client = new MqttClient(HOST, clentid,new MemoryPersistence());
        connect();

    }
    /**
     * 连接服务器
     */
    private void connect(){
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(false);
        options.setUserName(userName);
        options.setPassword(passWord.toCharArray());
        //设置超时时间
        options.setConnectionTimeout(10);
        //设置回话心跳时间
        options.setKeepAliveInterval(20);

        try {

            client.setCallback(new PushCallback());
            client.connect(options);

            topic11 = client.getTopic(TOPIC);

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

    }

    /**
     *
     * @param topic
     * @param message
     * @throws MqttPersistenceException
     * @throws MqttException
     */
    public void publish(MqttTopic topic,MqttMessage message) throws MqttException {

        MqttDeliveryToken token = topic.publish(message);
        token.waitForCompletion();
        System.out.println("message is published completely!");

    }

    /**
     *  启动入口
     * @param args
     * @throws MqttException
     */
    public static void main(String[] args) throws MqttException {
        MqttService server = new MqttService();

        server.message = new MqttMessage();
        server.message.setQos(1);
        server.message.setRetained(true);
        server.message.setPayload("哈哈哈哈".getBytes());
        server.publish(server.topic11 , server.message);
        System.out.println(server.message.isRetained() + "------ratained状态");
    }
}

客户端代码：

package com.example.dtest.mqtt;

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.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.util.concurrent.ScheduledExecutorService;

public class MyMqttClient {

    public static final String HOST = "tcp://0.0.0.0:61613";
    public static final String TOPIC = "topic11";
    private static final String clientid = "client11";
    private MqttClient client;
    private MqttConnectOptions options;
    private String userName = "admin";
    private String passWord = "password";

    private ScheduledExecutorService scheduler;

    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(true);

            // 设置连接的用户名
            options.setUserName(userName);

            // 设置连接的密码
            options.setPassword(passWord.toCharArray());

            // 设置超时时间 单位为秒
            options.setConnectionTimeout(10);
            // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，
            // 但这个方法并没有重连的机制
            options.setKeepAliveInterval(20);
            // 设置回调
            client.setCallback(new PushCallback());
            MqttTopic topic = client.getTopic(TOPIC);
            //setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息
            options.setWill(topic,"close".getBytes(), 2, true);

            client.connect(options);

            int[] Qos = {1};
            String[] topic1 = {TOPIC};
            client.subscribe(topic1,Qos);




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


    }

    public static void main(String[] args) {

        MyMqttClient client = new MyMqttClient();
        client.start();

    }



}

回滚代码：一般非阻塞方式，都是要使用回滚形式的：

package com.example.dtest.mqtt;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
public class PushCallback implements MqttCallback {
    @Override
    public void connectionLost(Throwable throwable) {
        //连接丢失后，一般在这里面进行重连
        System.out.println("连接断开，可以做重连");
    }

    //客户端回调
    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {

        // subscribe后得到的消息会执行到这里面
        System.out.println("接收消息主题 : "+topic);
        System.out.println("接收消息Qos : " +mqttMessage.getQos());
        System.out.println("接收消息内容 : " +new String(mqttMessage.getPayload()));

    }

    //服务端回调
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        System.out.println("deliveryComplete---------" );
    }
}