如何用 Eclipse Paho 开发 SpringBoot MQTT 客户端？

在物联网设备连接数突破300亿的时代背景下，MQTT协议凭借其轻量级、低带宽消耗和可靠的消息传输特性，已成为IoT领域的事实标准协议。本文将手把手教你基于Eclipse Paho和SpringBoot 2.5.15构建企业级MQTT客户端组件，通过注解驱动实现智能重连、消息路由等核心功能，帮助开发者快速搭建高可用的物联网通信平台。

JDK 8+（推荐JDK 11）
SpringBoot 2.5.15
Maven 3.6+
EMQX 5.0+（MQTT Broker）

<dependency>
  <groupId>org.eclipse.paho</groupId>
  <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
  <version>1.2.5</version>
</dependency>
<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-web</artifactId>
</dependency>

public class MqttConnectionManager implements MqttCallbackExtended {
  private MqttClient client;
  private MqttConnectOptions options;
  
  @PostConstruct
  public void init() throws MqttException {
    client = new MqttClient(serverURI, clientId, new MemoryPersistence());
    options = new MqttConnectOptions();
    options.setCleanSession(true);
    options.setAutomaticReconnect(true);
    options.setConnectionTimeout(10);
    client.connect(options);
  }
  
  // 实现连接状态回调方法
  @Override
  public void connectComplete(boolean reconnect, String serverURI) {
    System.out.println("连接状态：" + (reconnect ? "重连成功" : "首次连接"));
  }
}

在application.properties中添加：

mqtt.serverURI=tcp://127.0.0.1:1883
mqtt.clientId=springboot_client_${random.uuid}
mqtt.qosLevel=1
mqtt.keepAliveInterval=60

@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface MqttMessageListener {
  String topic();
  int qos() default 0;
}

public class MqttListenerProcessor implements BeanPostProcessor {
  @Autowired
  private MqttConnectionManager connectionManager;
  
  @Override
  public Object postProcessAfterInitialization(Object bean, String beanName) {
    Arrays.stream(bean.getClass().getMethods())
      .filter(method -> method.isAnnotationPresent(MqttMessageListener.class))
      .forEach(method -> {
        MqttMessageListener annotation = method.getAnnotation(MqttMessageListener.class);
        connectionManager.subscribe(annotation.topic(), annotation.qos(), message -> {
          method.invoke(bean, message.getPayload());
        });
      });
    return bean;
  }
}

public class SmartReconnectStrategy {
  private static final int MAX_RETRY = 10;
  private static final long BASE_DELAY = 1000;
  
  public void reconnect(MqttClient client) {
    int retryCount = 0;
    while (retryCount < MAX_RETRY) {
      try {
        Thread.sleep((long) (BASE_DELAY  Math.pow(2, retryCount)));
        client.reconnect();
        return;
      } catch (Exception e) {
        retryCount++;
      }
    }
    throw new MqttException("超过最大重试次数");
  }
}

public class MessageRouter {
  private Map<String, List<Consumer<byte[]>>> topicHandlers = new ConcurrentHashMap<>();
  
  public void addHandler(String topicFilter, Consumer<byte[]> handler) {
    topicHandlers.computeIfAbsent(topicFilter, k -> new ArrayList<>()).add(handler);
  }
  
  public void dispatch(String topic, byte[] payload) {
    topicHandlers.entrySet().stream()
      .filter(entry -> matchesTopic(topic, entry.getKey()))
      .forEach(entry -> entry.getValue().forEach(handler -> handler.accept(payload)));
  }
  
  private boolean matchesTopic(String actualTopic, String topicFilter) {
    // 实现MQTT通配符匹配逻辑
  }
}