MQTT 协议

package org.example;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import java.util.Arrays;


/**
 * MQTT 客户端演示
 */
public class MqttClientDemo {

  /** MQTT 客户端 */
  private IMqttClient mMqttClient;

  private final MqttCallback mqttCallback = new MqttCallback() {

    @Override
    public void connectionLost(Throwable throwable) {
      System.out.println("与服务器连接丢失, errorMsg = " + throwable.getMessage());
    }

    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) {
      System.out.println("收到服务器消息, topic = " + topic + ", payload = " + new String(mqttMessage.getPayload()));
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
      try {
        System.out.println("消息已传递完成且已收到所有确认, topic = " + Arrays.toString(token.getTopics()) + ", payload = " + new String(token.getMessage().getPayload()));
      } catch (MqttException e) {
        throw new RuntimeException(e);
      
    }
  };

  /**
   * 创建 MQTT 客户端
   * @param serverURI - 代理服务器的地址
   */
  public void createClient(String serverURI) {
      String clientId = MqttClient.generateClientId();
      MqttClientPersistence persistence = new MemoryPersistence();
      try {
        // serverURI - 服务器的地址。支持两种类型的连接 tcp:// 用于 TCP 连接，ssl:// 用于受 SSL/TLS 保护的 TCP 连接。
        // clientId - 连接到代理时使用的唯一客户端 ID 字符串。serverURI 参数通常与 clientId 参数一起使用以形成密钥，用于在传递消息时存储和引用消息。
        // persistence - 用于存储传输中的消息的持久化类。如果为空，则使用默认的持久化机制。
        mMqttClient = new MqttClient(serverURI, clientId, persistence);
        System.out.println("创建 MQTT 客户端: serverURI = " + serverURI + ", clientId = " + clientId);
      } catch (MqttException e) {
        e.printStackTrace();
      }
  }

  /**
   * 通过用户名密码连接到 MQTT 代理服务器
   * @param username - 用户名
   * @param password - 密码
   */
  public void connect(String username, String password) {
    MqttConnectOptions options = new MqttConnectOptions();
    options.setUserName(username);
    options.setPassword(password.toCharArray());
    // 返回如果连接丢失，客户端是否会自动尝试重新连接到服务器
    options.setAutomaticReconnect(true);
    // 设置客户端和服务器是否应在重新启动和重新连接时记住状态。
    options.setCleanSession(true);
    // 设置回调用于异步发生的事件，如收到新消息、消息发送成功、与服务器连接丢失
    mMqttClient.setCallback(mqttCallback);
    try {
      mMqttClient.connect(options);
      System.out.println("通过用户名密码连接到 MQTT 代理服务器: username = " + username + ", password = " + password);
    } catch (MqttException e) {
      e.printStackTrace();
    }
  }

  /**
   * 发布消息
   * @param topic - 主题
   * @param payload - 负载，实际使用的消息内容
   * @param qos - 消息发布的服务质量，QoS0 - 0，QoS1 - 1，QoS2 - 2
   * @param retain - 发布消息是否应由代理服务器保留
   * @return boolean - 消息是否发布成功
   */
  public boolean publish(String topic, String payload, int qos, boolean retain) {
      MqttMessage mqttMessage = new MqttMessage();
      mqttMessage.setPayload(payload.getBytes());
      mqttMessage.setQos(qos);
      mqttMessage.setRetained(retain);
      try {
        mMqttClient.publish(topic, mqttMessage);
        System.out.println("发布消息: topic = " + topic + ", payload = " + payload);
        return true;
      } catch (MqttException e) {
        e.printStackTrace();
        return false;
      }
  }

  /**
   * 订阅一个主题
   * @param topic - 要订阅的主题，可以包含通配符
   * @param qos - 订阅的最高服务质量，以较低服务质量发布的消息将以发布的 QoS 接收
   */
  public void subscribe(String topic, int qos) {
    try {
      mMqttClient.subscribe(topic, qos);
      System.out.println("订阅一个主题: topic = " + topic + ", qos = " + qos);
    } catch (MqttException e) {
      e.printStackTrace();
    }
  }

  /**
   * 退订一个主题
   * @param topic - 要退订的主题，必须与订阅中指定的主题匹配
   */
  public void unsubscribe(String topic) {
    try {
      mMqttClient.unsubscribe(topic);
      System.out.println("退订一个主题: topic = " + topic);
    } catch (MqttException e) {
      e.printStackTrace();
    }
  }

  /**
   * 断开与服务器的连接
   */
  public void disConnect() {
    try {
      mMqttClient.disconnect();
      System.out.println("断开与服务器的连接");
    } catch (MqttException e) {
      e.printStackTrace();
    }
  }

  /**
   * 重新连接服务器
   */
  public void reConnect() {
    try {
      mMqttClient.reconnect();
      System.out.println("重新连接服务器");
    } catch (MqttException e) {
      e.printStackTrace();
    }
  }
}

MqttClientDemo client = new MqttClientDemo();
client.createClient("tcp://localhost:1883");
client.connect("user", "123456");
new Thread() {
  @Override
  public void run() {
    super.run();
    try {
      Thread.sleep(5000);
      client.subscribe("test", 0);
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    }
  }
}.start();

MqttClientDemo client = new MqttClientDemo();
client.createClient("tcp://localhost:1883");
client.connect("user", "123456");
new Thread() {
  @Override
  public void run() {
    super.run();
    try {
      Thread.sleep(5000);
      client.subscribe("test", 0);
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    }
  }
}.start();

try {
  InputStream is = mSocket.getInputStream();
  OutputStream os = mSocket.getOutputStream();
  // ...忽略部分变量定义...
  while (true) {
    read = is.read(buffer, 0, 1); // 读取报文第 1 个字节
    length = BitUtils.decodeRemainingLength(is); // 解码固定报头中的剩余长度字段
    type = BitUtils.getHigh4bit(buffer[0]); // 读取报文第 1 个字节中的高 4 位， 即报文类型
    flag = BitUtils.getLow4bit(buffer[0]); // 读取报文第 1 个字节中的低 4 位， 即报文类型标记
    System.out.println("\n固定报头：类型 = " + type + ", flag = " + flag + ", length = " + length);
  }
} catch (IOException e) {
  e.printStackTrace();
}

if (type == 1) {
  // CONNECT - 客户向服务器发送的连接请求

  // 回复连接确认报文 CONNACK，固定报头 2 字节 + 可变报头 2 字节
  // 可变报头内容为连接确认标记 0 和连接确认码 0
  os.write(new byte[] { 0x20, 0x02, 0x00, 0x00 });
  os.flush();
} else if (type == 3) {
  // PUBLISH - 发布报文，忽略 QoS

  // qos = 0， 无需回复 PUBACK
} else if (type == 8 && flag == 2) {
  // SUBSCRIBE - 订阅主题

  // 回复订阅主题报文 SUBACK, 总长度为 固定报头 2 字节 + 可变报头 2 字节 + 1字节 * 订阅主题过滤器列表大小
  byte[] subacks = new byte[ 2 + 2 + topics.size() ];
  // 固定报头 2 字节
  // 由于 java 中没有无符号类型，所以 0x90 超出了 byte 的取值范围，强转处理即可
  subacks[0] = (byte)0x90; // 类型：SUBACK， 标记： 0
  subacks[1] = (byte)(2 + topics.size()); // 剩余长度 = 可变报头 2 字节 + 主题过滤器列表大小 * 1字节, 此处忽略剩余长度大于 127 的情况
  // 可变报头 2 字节 - 报文标识符 = 0x0001
  subacks[2] = 0x00;
  subacks[3] = 0x01;
  // 有效载荷
  for(int i = 4; i < subacks.length; i ++) {
    // 简单处理，统一设置回复主题过滤器 Qos = 0
    subacks[i] = 0x00;
  }
  // 回复 SUBACK
  os.write(subacks);
  os.flush();

} else if (type == 12) {
  // PINGREQ – 心跳请求
  System.out.println("======== PINGREQ 心跳请求 ========");
  // 回复心跳请求报文 PINGRESP
  os.write(new byte[] { (byte)0xD0, 0x00 });
}

package org.example;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;

/**
 * MQTT 代理服务器演示
 */
public class MqttBrokerDemo {

    private static final int port = 1883;
    private HashMap<Socket, String> clientMap = new HashMap<Socket, String>();
    private HashMap<String, ArrayList<Socket>> topics = new HashMap<String, ArrayList<Socket>>();

    public static void main(String[] args) {
        new MqttBrokerDemo().run();
    }


    public void run() {
        try {
            ServerSocket serverSocket = new ServerSocket(port);
            System.out.println("MQTT Broker started on port " + port);
            while (true) {
                Socket clientSocket = serverSocket.accept();
                System.out.println("\n有新连接 getInetAddress：" + clientSocket.getInetAddress());
                new MqttBrokerServer(clientSocket, new CallBack() {
                    /**
                     * 客户端连接请求回调
                     * @param socket - 当前客户端 Socket 对象
                     * @param clientId - 客户端标识符
                     */
                    @Override
                    public void connect(Socket socket, String clientId) {
                        clientMap.put(socket, clientId);
                    }

                    /**
                     * 客户端发布请求回调
                     * @param socket - 当前客户端 Socket 对象
                     * @param topic - 主题
                     * @param bytes - publish 数据包，用于转发给目标客户端
                     */
                    @Override
                    public void publish(Socket socket, String topic, byte[] bytes) {
                        ArrayList<Socket> array = topics.get(topic);
                        if (array == null) return;
                        try {
                            for (Socket temp : array) {
                                OutputStream os = temp.getOutputStream();
                                os.write(bytes);
                                os.flush();
                            }
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }

                    /**
                     * 客户端订阅请求回调
                     * @param socket - 当前客户端 Socket 对象
                     * @param topicList - 主题过滤器列表
                     */
                    @Override
                    public void subscribe(Socket socket, ArrayList<String> topicList) {
                        if (topicList == null) return;
                        for (String topic : topicList) {
                            ArrayList<Socket> sockets = topics.get(topic);
                            if (sockets == null) {
                                sockets = new ArrayList<Socket>();
                                sockets.add(socket);
                                topics.put(topic, sockets);
                            } else {
                                sockets.add(socket);
                                topics.put(topic, sockets);
                            }
                        }
                    }
                }).start();
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    static class MqttBrokerServer extends Thread {

        protected Socket mSocket;
        protected CallBack mCallBack;

        public MqttBrokerServer(Socket socket, CallBack callBack) {
            this.mSocket = socket;
            this.mCallBack = callBack;
        }

        @Override
        public void run() {
            super.run();
            try {
                InputStream is = mSocket.getInputStream();
                OutputStream os = mSocket.getOutputStream();
                byte[] buffer = new byte[1024];
                int type = 0;
                int flag = 0;
                int length = 0;
                int read = 0;
                long skipLen = 0;
                while (true) {
                    read = is.read(buffer, 0 ,1);
                    length = BitUtils.decodeRemainingLength(is);
                    type = BitUtils.getHigh4bit(buffer[0]);
                    flag = BitUtils.getLow4bit(buffer[0]);
                    System.out.println("\n固定报头：类型 = " + type + ", flag = "+ flag +", length = " + length);

                    if (type == 1) {
                        // CONNECT - 客户向服务器发送的连接请求
                        // 读取 CONNECT 报文的可变报头
                        // 读取协议名长度
                        read = is.read(buffer, 0 ,2);
                        int protocolLen = BitUtils.getIntBy2ByteData(buffer[0], buffer[1]);
                        // 读取协议名
                        read = is.read(buffer, 0 , protocolLen);
                        String protocolName = new String(Arrays.copyOfRange(buffer, 0, protocolLen), StandardCharsets.UTF_8);
                        // 读取部分连接标记
                        skipLen = is.skip(1);
                        read = is.read(buffer, 0 ,1);
                        int userNameFlag = BitUtils.getBit(buffer[0], 7); // 用户名标志
                        int pwdFlag = BitUtils.getBit(buffer[0], 6); // 密码标志
                        // 读取心跳连接间隔
                        read = is.read(buffer, 0 ,2);
                        int heartbeat =  BitUtils.getIntBy2ByteData(buffer[0], buffer[1]);
                        // 读取 clientId
                        read = is.read(buffer, 0, 2);
                        int clientIdLen = BitUtils.getIntBy2ByteData(buffer[0], buffer[1]);
                        read = is.read(buffer, 0 , clientIdLen);
                        String clientId = new String(Arrays.copyOfRange(buffer, 0, clientIdLen), StandardCharsets.UTF_8);
                        // 读取 userName
                        read = is.read(buffer, 0, 2);
                        int userNameLen = BitUtils.getIntBy2ByteData(buffer[0], buffer[1]);
                        read = is.read(buffer, 0 , userNameLen);
                        String userName = new String(Arrays.copyOfRange(buffer, 0, userNameLen), StandardCharsets.UTF_8);
                        // 读取 password
                        read = is.read(buffer, 0, 2);
                        int passwordLen = BitUtils.getIntBy2ByteData(buffer[0], buffer[1]);
                        read = is.read(buffer, 0 , passwordLen);
                        String password = new String(Arrays.copyOfRange(buffer, 0, passwordLen), StandardCharsets.UTF_8);

                        System.out.println("======== CONNECT 连接请求 ========");
                        System.out.println("protocolLen = " + protocolLen + ", protocolName = " + protocolName + ", userNameFlag = " + userNameFlag + ", pwdFlag = " + pwdFlag + ", heartbeat = " + heartbeat + "s");
                        System.out.println("clientId = " + clientId + ", userName = " + userName + ", password = " + password);

                        mCallBack.connect(mSocket, clientId);

                        // 回复连接确认报文 CONNACK，固定报头 2 字节 + 可变报头 2 字节
                        // 可变报头内容为连接确认标记 0 和连接确认码 0
                        os.write(new byte[] { 0x20, 0x02, 0x00, 0x00 });
                        os.flush();

                    } else if (type == 3) {
                        // PUBLISH - 发布报文，忽略 QoS
                        ArrayList<Byte> byteList = new ArrayList<>();
                        byteList.add(buffer[0]);
                        for (byte b : BitUtils.encodeRemainingLength(length)) {
                            byteList.add(b);
                        }
                        int messageLength = length;

                        // 读取 topic
                        read = is.read(buffer, 0, 2);
                        int topicLen = BitUtils.getIntBy2ByteData(buffer[0], buffer[1]);
                        byteList.add(buffer[0]);
                        byteList.add(buffer[1]);
                        read = is.read(buffer, 0 , topicLen);
                        byte[] topicByte = Arrays.copyOfRange(buffer, 0, topicLen);
                        String topic = new String(topicByte, StandardCharsets.UTF_8);
                        for (byte b : topicByte) { byteList.add(b); }

                        // Qos = 0, 无报文标识符
                        // 读取 message
                        messageLength = messageLength - 2 - topicLen;
                        read = is.read(buffer, 0 , messageLength);
                        byte[] msgByte = Arrays.copyOfRange(buffer, 0, messageLength);
                        String message = new String(msgByte, StandardCharsets.UTF_8);
                        for (byte b : msgByte) { byteList.add(b); }

                        System.out.println("======== PUBLISH 发布请求 ========");
                        System.out.println("topic = " + topic + ", message = " + message);

                        byte[] bytes = new byte[byteList.size()];
                        for(int i = 0; i < bytes.length; i ++) {
                            bytes[i] = byteList.get(i);
                        }
                        mCallBack.publish(mSocket, topic, bytes);
                        // qos = 0， 无需回复 PUBACK

                    } else if (type == 8 && flag == 2) {
                        // SUBSCRIBE - 订阅主题
                        int messageLength = length;
                        // 跳过可变报头 - 报文标识符
                        skipLen = is.skip(2);
                        messageLength -= skipLen;

                        // 读取 topic
                        ArrayList<String> topics = new ArrayList<String>();
                        while(messageLength > 0) {
                            read = is.read(buffer, 0 ,2);
                            int topicLen = BitUtils.getIntBy2ByteData(buffer[0], buffer[1]);
                            read = is.read(buffer, 0 , topicLen);
                            String topic = new String(Arrays.copyOfRange(buffer, 0, topicLen), StandardCharsets.UTF_8);
                            topics.add(topic);
                            // Qos 暂时忽略
                            read = is.read(buffer, 0 , 1);
                            messageLength -= (topicLen + 3);
                        }
                        System.out.println("======== SUBSCRIBE 订阅请求 ========");
                        System.out.println("topics = " + Arrays.toString(topics.toArray()));

                        mCallBack.subscribe(mSocket, topics);

                        // 回复订阅主题报文 SUBACK, 总长度为 固定报头 2 字节 + 可变报头 2 字节 + 1字节 * 订阅主题过滤器列表大小
                        byte[] subacks = new byte[ 2 + 2 + topics.size() ];
                        // 固定报头 2 字节
                        // 由于 java 中没有无符号类型，所以 0x90 超出了 byte 的取值范围，强转处理即可
                        subacks[0] = (byte)0x90; // 类型：SUBACK， 标记： 0
                        subacks[1] = (byte)(2 + topics.size()); // 剩余长度 = 可变报头 2 字节 + 主题过滤器列表大小 * 1字节, 此处忽略剩余长度大于 127 的情况
                        // 可变报头 2 字节 - 报文标识符 = 0x0001
                        subacks[2] = 0x00;
                        subacks[3] = 0x01;
                        // 有效载荷
                        for(int i = 4; i < subacks.length; i ++) {
                            // 简单处理，统一设置回复主题过滤器 Qos = 0
                            subacks[i] = 0x00;
                        }
                        // 回复 SUBACK
                        os.write(subacks);
                        os.flush();

                    } else if (type == 12) {
                        // PINGREQ – 心跳请求
                        System.out.println("======== PINGREQ 心跳请求 ========");
                        // 回复心跳请求报文 PINGRESP
                        os.write(new byte[] { (byte)0xD0, 0x00 });
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    interface CallBack {
        public void connect(Socket socket, String clientId);

        public void publish(Socket socket, String topic, byte[] bytes);

        public void subscribe(Socket socket, ArrayList<String> topicList);
    }

}

package org.example;

import java.io.IOException;
import java.io.InputStream;

public class BitUtils {

    /**
     * 解码 MQTT 固定报头剩余长度
     * @param is - InputStream
     * @return int
     */
    public static int decodeRemainingLength(InputStream is) throws IOException {
        byte[] buffer = new byte[1];
        int length = 0;
        int multiplier = 1;
        int max = 128 * 128 * 128;
        int read = 0;
        do {
            read = is.read(buffer, 0 ,1);
            if (read == -1) break;
            length += (buffer[0] & 0x7f) * multiplier;
            multiplier *= 128;
            if (multiplier > max) break;
        } while ((buffer[0] & 0x80) != 0);
        return length;
    }

    /**
     * 编码 MQTT 固定报头剩余长度为 byte 数组
     * @param length - 固定报头剩余长度
     * @return byte[]
     */
    public static byte[] encodeRemainingLength(int length) {
        byte[] temp = new byte[4];
        int index = 0;
        int number = length;
        do {
            temp[index] = (byte)(number % 128);
            number /= 128;
            if (number > 0) {
                // 向最高位添加标志 1
                temp[index] = (byte)(temp[index] | 128);
            }
            index ++;
        } while (number > 0);

        byte[] bytes = new byte[index];
        System.arraycopy(temp, 0, bytes, 0, bytes.length);
        return bytes;
    }


    /**
     * 获取 Byte 数据的高 4 位
     * @param data - byte 数据
     * @return int
     */
    public static int getHigh4bit(byte data) {
        return (data & 0xF0) >> 4;
    }

    /**
     * 获取 byte 数据的低 4 位
     * @param data - byte 数据
     * @return int
     */
    public static int getLow4bit(byte data) {
        return data & 0x0F;
    }

    /**
     * 通过字节数据中的某一二进制位
     * @param data - 字节数据
     * @param index - 第几个二进制位, 从 0 开始
     * @return int
     */
    public static int getBit(byte data, int index) {
        return ((1 << index) & data) >> index;
    }


    /**
     * 通过高字节和低字节还原数据值
     * @param high - 据高字节
     * @param low - 数据低字节
     * @return int
     */
    public static int getIntBy2ByteData(byte high, byte low) {
        return ((high & 0x00FF) << 8) | low;
    }

}