基于springboot设计实现mqtt协议接入，示范代码包括发送端，订阅端，网关借口

MQTT是一种轻量级的物联网协议，可以实现低功耗设备之间的通信。下面是基于Spring Boot实现MQTT协议接入的示范代码，包括发送端、订阅端和网关接口。

1. 发送端代码示例

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.stereotype.Component;

@Component
public class MqttSender implements MqttCallback {

    private final String broker = "tcp://localhost:1883";
    private final String clientId = "sender";
    private final String topic = "hello";
    private MqttAsyncClient client;

    public MqttSender() throws MqttException {
        client = new MqttAsyncClient(broker, clientId);
        client.setCallback(this);
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true);
        IMqttToken token = client.connect(options);
        token.waitForCompletion();
    }

    public void send(String message) throws MqttException {
        MqttMessage mqttMessage = new MqttMessage(message.getBytes());
        mqttMessage.setQos(1);
        client.publish(topic, mqttMessage);
    }

    @Override
    public void connectionLost(Throwable cause) {
        // 连接丢失时的处理
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        // 收到消息时的处理
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        // 消息发送完成时的处理
    }
}

2. 订阅端代码示例

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttCallback;
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;
import org.springframework.stereotype.Component;

@Component
public class MqttSubscriber implements MqttCallback {

    private final String broker = "tcp://localhost:1883";
    private final String clientId = "subscriber";
    private final String topic = "hello";
    private MqttAsyncClient client;

    public MqttSubscriber() throws MqttException {
        client = new MqttAsyncClient(broker, clientId, new MemoryPersistence());
        client.setCallback(this);
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true);
        IMqttToken token = client.connect(options);
        token.waitForCompletion();
        client.subscribe(topic, 1);
    }

    @Override
    public void connectionLost(Throwable cause) {
        // 连接丢失时的处理
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        // 收到消息时的处理
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        // 消息发送完成时的处理
    }
}

3. 网关接口代码示例

import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.ResponseBody;

@Controller
public class MqttGateway {

    @Autowired
    private MqttSender sender;

    @PostMapping("/mqtt/send")
    @ResponseBody
    public String send(@RequestBody String message) throws MqttException {
        sender.send(message);
        return "Success";
    }

    @PostMapping("/mqtt/subscribe")
    @ResponseBody
    public String subscribe(@RequestBody String message) {
        return "Success";
    }
}

这些示例代码实现了MQTT协议的发送和订阅功能，通过网关接口可以让其他系统调用。注意在使用示例代码时需要根据具体情况修改Broker地址和端口号等参数。