📡 Java + MQTT 即时通讯全方案：从协议原理到集群部署

public enum MQTTQoS {
    /**
     * 最多一次传输 (QoS 0)
     * 消息可能丢失，但不重传
     */
    AT_MOST_ONCE(0),
    
    /**
     * 至少一次传输 (QoS 1)
     * 消息不会丢失，但可能重复
     */
    AT_LEAST_ONCE(1),
    
    /**
     * 恰好一次传输 (QoS 2)
     * 消息不会丢失且不会重复
     */
    EXACTLY_ONCE(2);
    
    private final int value;
    
    MQTTQoS(int value) {
        this.value = value;
    }
    
    public int getValue() {
        return value;
    }
}

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import com.fasterxml.jackson.databind.ObjectMapper;

/**
 * MQTT客户端管理类
 * 提供连接管理、消息发布和订阅功能
 */
public class MQTTClientManager implements MqttCallback {
    
    private MqttClient mqttClient;
    private MqttConnectOptions connectOptions;
    private final String brokerUrl;
    private final String clientId;
    private final MemoryPersistence persistence;
    private final ObjectMapper objectMapper;
    
    private static final int CONNECTION_TIMEOUT = 30;
    private static final int KEEP_ALIVE_INTERVAL = 60;
    
    /**
     * 构造函数
     * @param brokerUrl MQTT代理地址
     * @param clientId 客户端ID
     */
    public MQTTClientManager(String brokerUrl, String clientId) {
        this.brokerUrl = brokerUrl;
        this.clientId = clientId;
        this.persistence = new MemoryPersistence();
        this.objectMapper = new ObjectMapper();
        initializeConnectOptions();
    }
    
    /**
     * 初始化连接选项
     */
    private void initializeConnectOptions() {
        connectOptions = new MqttConnectOptions();
        connectOptions.setCleanSession(true); // 清除会话
        connectOptions.setConnectionTimeout(CONNECTION_TIMEOUT);
        connectOptions.setKeepAliveInterval(KEEP_ALIVE_INTERVAL);
        connectOptions.setAutomaticReconnect(true); // 自动重连
        
        // 设置遗嘱消息
        String willMessage = "{\"type\": \"offline\", \"clientId\": \"" + clientId + "\"}";
        connectOptions.setWill("clients/" + clientId + "/status", 
            willMessage.getBytes(), 1, true);
    }
    
    /**
     * 连接到MQTT代理
     * @param username 用户名
     * @param password 密码
     * @throws MqttException 连接异常
     */
    public void connect(String username, char[] password) throws MqttException {
        if (username != null && password != null) {
            connectOptions.setUserName(username);
            connectOptions.setPassword(password);
        }
        
        try {
            mqttClient = new MqttClient(brokerUrl, clientId, persistence);
            mqttClient.setCallback(this);
            mqttClient.connect(connectOptions);
            
            // 发布在线状态
            publish("clients/" + clientId + "/status", 
                "{\"type\": \"online\", \"clientId\": \"" + clientId + "\"}", 1, true);
            
            System.out.println("MQTT客户端连接成功: " + clientId);
        } catch (MqttException e) {
            System.err.println("MQTT连接失败: " + e.getMessage());
            throw e;
        }
    }
    
    /**
     * 发布消息
     * @param topic 主题
     * @param payload 消息内容
     * @param qos 服务质量等级
     * @param retained 是否保留消息
     * @throws MqttException 发布异常
     */
    public void publish(String topic, String payload, int qos, boolean retained) 
            throws MqttException {
        if (mqttClient == null || !mqttClient.isConnected()) {
            throw new MqttException(MqttException.REASON_CODE_CLIENT_NOT_CONNECTED);
        }
        
        MqttMessage message = new MqttMessage(payload.getBytes());
        message.setQos(qos);
        message.setRetained(retained);
        
        mqttClient.publish(topic, message);
    }
    
    /**
     * 发布JSON消息
     * @param topic 主题
     * @param payloadObject 消息对象
     * @param qos 服务质量等级
     * @param retained 是否保留消息
     * @throws MqttException 发布异常
     */
    public void publishJson(String topic, Object payloadObject, int qos, boolean retained) 
            throws MqttException {
        try {
            String payload = objectMapper.writeValueAsString(payloadObject);
            publish(topic, payload, qos, retained);
        } catch (Exception e) {
            throw new MqttException(MqttException.REASON_CODE_UNEXPECTED_ERROR);
        }
    }
    
    /**
     * 订阅主题
     * @param topic 主题名称
     * @param qos 服务质量等级
     * @throws MqttException 订阅异常
     */
    public void subscribe(String topic, int qos) throws MqttException {
        if (mqttClient == null || !mqttClient.isConnected()) {
            throw new MqttException(MqttException.REASON_CODE_CLIENT_NOT_CONNECTED);
        }
        
        mqttClient.subscribe(topic, qos);
        System.out.println("已订阅主题: " + topic);
    }
    
    /**
     * 取消订阅
     * @param topic 主题名称
     * @throws MqttException 取消订阅异常
     */
    public void unsubscribe(String topic) throws MqttException {
        if (mqttClient != null && mqttClient.isConnected()) {
            mqttClient.unsubscribe(topic);
        }
    }
    
    /**
     * 断开连接
     * @throws MqttException 断开异常
     */
    public void disconnect() throws MqttException {
        if (mqttClient != null && mqttClient.isConnected()) {
            // 发布离线状态
            publish("clients/" + clientId + "/status", 
                "{\"type\": \"offline\", \"clientId\": \"" + clientId + "\"}", 1, true);
            
            mqttClient.disconnect();
            System.out.println("MQTT客户端已断开连接");
        }
    }
    
    // ========== MqttCallback 接口实现 ==========
    
    @Override
    public void connectionLost(Throwable cause) {
        System.err.println("MQTT连接丢失: " + cause.getMessage());
        // 实现重连逻辑
        attemptReconnect();
    }
    
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        String payload = new String(message.getPayload());
        System.out.println("收到消息 - 主题: " + topic + ", 内容: " + payload);
        
        // 处理消息
        handleIncomingMessage(topic, payload);
    }
    
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        // 消息发布完成回调
        System.out.println("消息发布完成: " + token.getMessageId());
    }
    
    /**
     * 处理接收到的消息
     * @param topic 主题
     * @param payload 消息内容
     */
    private void handleIncomingMessage(String topic, String payload) {
        try {
            // 根据主题模式处理不同消息
            if (topic.startsWith("chat/")) {
                handleChatMessage(topic, payload);
            } else if (topic.startsWith("system/")) {
                handleSystemMessage(topic, payload);
            }
            // 添加其他主题处理逻辑...
        } catch (Exception e) {
            System.err.println("消息处理错误: " + e.getMessage());
        }
    }
    
    /**
     * 处理聊天消息
     */
    private void handleChatMessage(String topic, String payload) {
        try {
            ChatMessage chatMessage = objectMapper.readValue(payload, ChatMessage.class);
            System.out.println("收到聊天消息 from " + chatMessage.getFrom() + 
                             ": " + chatMessage.getContent());
        } catch (Exception e) {
            System.err.println("聊天消息解析失败: " + e.getMessage());
        }
    }
    
    /**
     * 尝试重新连接
     */
    private void attemptReconnect() {
        int maxAttempts = 5;
        int attempt = 0;
        
        while (attempt < maxAttempts) {
            try {
                Thread.sleep(5000 * (attempt + 1)); // 指数退避
                if (!mqttClient.isConnected()) {
                    mqttClient.reconnect();
                    System.out.println("重连成功");
                    return;
                }
            } catch (Exception e) {
                attempt++;
                System.err.println("重连尝试 " + attempt + " 失败: " + e.getMessage());
            }
        }
        
        System.err.println("达到最大重连次数，连接失败");
    }
}

/**
 * 聊天消息实体类
 */
@Data
@NoArgsConstructor
@AllArgsConstructor
class ChatMessage {
    private String from;
    private String to;
    private String content;
    private long timestamp;
    private MessageType type;
    
    public enum MessageType {
        TEXT, IMAGE, FILE, SYSTEM
    }
}

/**
 * MQTT消息处理器接口
 */
public interface MessageHandler {
    
    /**
     * 处理消息
     * @param topic 消息主题
     * @param message 消息内容
     */
    void handleMessage(String topic, MqttMessage message);
    
    /**
     * 支持的主题模式
     * @return 主题模式数组
     */
    String[] getSupportedTopics();
}

/**
 * 聊天消息处理器
 */
public class ChatMessageHandler implements MessageHandler {
    
    private final ObjectMapper objectMapper = new ObjectMapper();
    
    @Override
    public void handleMessage(String topic, MqttMessage message) {
        try {
            ChatMessage chatMessage = objectMapper.readValue(
                new String(message.getPayload()), ChatMessage.class);
            
            // 根据消息类型处理
            switch (chatMessage.getType()) {
                case TEXT:
                    processTextMessage(chatMessage);
                    break;
                case IMAGE:
                    processImageMessage(chatMessage);
                    break;
                case FILE:
                    processFileMessage(chatMessage);
                    break;
            }
        } catch (Exception e) {
            System.err.println("处理聊天消息失败: " + e.getMessage());
        }
    }
    
    @Override
    public String[] getSupportedTopics() {
        return new String[]{"chat/#", "group/+/chat"};
    }
    
    private void processTextMessage(ChatMessage message) {
        System.out.println("处理文本消息: " + message.getContent());
        // 实现具体的消息处理逻辑
    }
    
    private void processImageMessage(ChatMessage message) {
        System.out.println("处理图片消息");
        // 实现图片消息处理
    }
    
    private void processFileMessage(ChatMessage message) {
        System.out.println("处理文件消息");
        // 实现文件消息处理
    }
}

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;

@Configuration
public class MQTTConfig {
    
    @Value("${mqtt.broker.url}")
    private String brokerUrl;
    
    @Value("${mqtt.client.id}")
    private String clientId;
    
    @Value("${mqtt.username}")
    private String username;
    
    @Value("${mqtt.password}")
    private String password;
    
    /**
     * MQTT连接选项Bean
     */
    @Bean
    public MqttConnectOptions mqttConnectOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[]{brokerUrl});
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setCleanSession(true);
        options.setAutomaticReconnect(true);
        options.setConnectionTimeout(30);
        options.setKeepAliveInterval(60);
        options.setMaxInflight(1000);
        
        return options;
    }
    
    /**
     * MQTT客户端工厂
     */
    @Bean
    public MqttClientFactory mqttClientFactory(MqttConnectOptions options) {
        DefaultMqttClientFactory factory = new DefaultMqttClientFactory();
        factory.setConnectionOptions(options);
        return factory;
    }
}

/**
 * MQTT消息监听器容器配置
 */
@Configuration
@EnableMqtt
public class MQTTListenerConfig extends AbstractMqttMessageDrivenChannelAdapter {
    
    @Autowired
    private MessageHandler[] messageHandlers;
    
    @Bean
    public MqttPahoMessageDrivenChannelAdapter mqttInbound(
            MqttClientFactory factory, 
            @Value("${mqtt.client.id}") String clientId) {
        
        MqttPahoMessageDrivenChannelAdapter adapter = 
            new MqttPahoMessageDrivenChannelAdapter(clientId + "-inbound", factory);
        
        // 动态添加所有处理器支持的主题
        for (MessageHandler handler : messageHandlers) {
            for (String topic : handler.getSupportedTopics()) {
                adapter.addTopic(topic, 1); // QoS 1
            }
        }
        
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        
        return adapter;
    }
    
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public void handleMessage(Message<?> message) {
        MqttHeaders headers = message.getHeaders();
        String topic = headers.get(MqttHeaders.RECEIVED_TOPIC).toString();
        byte[] payload = (byte[]) message.getPayload();
        
        // 分发消息到对应的处理器
        for (MessageHandler handler : messageHandlers) {
            if (matchesTopic(handler.getSupportedTopics(), topic)) {
                MqttMessage mqttMessage = new MqttMessage(payload);
                handler.handleMessage(topic, mqttMessage);
            }
        }
    }
    
    private boolean matchesTopic(String[] patterns, String topic) {
        for (String pattern : patterns) {
            if (topic.matches(pattern.replace("+", "[^/]+").replace("#", ".+"))) {
                return true;
            }
        }
        return false;
    }
}

@Service
@Slf4j
public class MQTTMessageService {
    
    @Autowired
    private MqttClientFactory mqttClientFactory;
    
    @Value("${mqtt.client.id}")
    private String clientId;
    
    private MqttTemplate mqttTemplate;
    
    @PostConstruct
    public void init() {
        this.mqttTemplate = new MqttTemplate(mqttClientFactory, clientId + "-outbound");
    }
    
    /**
     * 发送聊天消息
     */
    public void sendChatMessage(String from, String to, String content) {
        ChatMessage message = new ChatMessage(from, to, content, 
            System.currentTimeMillis(), ChatMessage.MessageType.TEXT);
        
        try {
            String topic = "chat/" + to;
            mqttTemplate.send(topic, message, 1, false);
            log.info("消息已发送到: {}", topic);
        } catch (Exception e) {
            log.error("发送消息失败", e);
            throw new MessageSendException("消息发送失败", e);
        }
    }
    
    /**
     * 发送群组消息
     */
    public void sendGroupMessage(String from, String groupId, String content) {
        ChatMessage message = new ChatMessage(from, groupId, content, 
            System.currentTimeMillis(), ChatMessage.MessageType.TEXT);
        
        try {
            String topic = "group/" + groupId + "/chat";
            mqttTemplate.send(topic, message, 1, false);
            log.info("群组消息已发送到: {}", topic);
        } catch (Exception e) {
            log.error("发送群组消息失败", e);
            throw new MessageSendException("群组消息发送失败", e);
        }
    }
    
    /**
     * 发送保留消息（用于状态同步）
     */
    public void sendRetainedMessage(String topic, Object payload) {
        try {
            mqttTemplate.send(topic, payload, 1, true);
            log.info("保留消息已发送到: {}", topic);
        } catch (Exception e) {
            log.error("发送保留消息失败", e);
        }
    }
}

/**
 * 自定义MQTT模板
 */
@Component
@Slf4j
public class MqttTemplate {
    
    private final MqttClientFactory clientFactory;
    private final String clientId;
    private MqttAsyncClient mqttClient;
    
    public MqttTemplate(MqttClientFactory clientFactory, String clientId) {
        this.clientFactory = clientFactory;
        this.clientId = clientId;
        initializeClient();
    }
    
    private void initializeClient() {
        try {
            MqttConnectOptions options = clientFactory.getConnectionOptions();
            mqttClient = new MqttAsyncClient(options.getServerURIs()[0], clientId, 
                new MemoryPersistence());
            
            mqttClient.connect(options).waitForCompletion();
            log.info("MQTT模板客户端连接成功");
        } catch (Exception e) {
            log.error("MQTT模板客户端初始化失败", e);
        }
    }
    
    public void send(String topic, Object payload, int qos, boolean retained) {
        try {
            String jsonPayload = new ObjectMapper().writeValueAsString(payload);
            MqttMessage message = new MqttMessage(jsonPayload.getBytes());
            message.setQos(qos);
            message.setRetained(retained);
            
            mqttClient.publish(topic, message);
        } catch (Exception e) {
            log.error("发送MQTT消息失败", e);
            throw new RuntimeException("消息发送失败", e);
        }
    }
}

/**
 * 可靠消息服务
 * 保证消息的可靠投递
 */
@Service
@Slf4j
public class ReliableMessageService {
    
    @Autowired
    private MessageStoreRepository messageStoreRepository;
    
    @Autowired
    private MqttTemplate mqttTemplate;
    
    private final ScheduledExecutorService retryScheduler = 
        Executors.newSingleThreadScheduledExecutor();
    
    /**
     * 发送可靠消息（QoS 2模拟）
     */
    public void sendReliableMessage(String topic, Object payload, String messageId) {
        // 存储消息到数据库
        MessageStore messageStore = new MessageStore(messageId, topic, payload, 
            MessageStatus.PENDING, System.currentTimeMillis());
        
        messageStoreRepository.save(messageStore);
        
        // 发送消息
        try {
            mqttTemplate.send(topic, new MessageWrapper(payload, messageId), 1, false);
            
            // 启动确认超时检查
            scheduleConfirmationCheck(messageId);
        } catch (Exception e) {
            messageStore.setStatus(MessageStatus.FAILED);
            messageStoreRepository.save(messageStore);
            log.error("可靠消息发送失败", e);
        }
    }
    
    /**
     * 处理消息确认
     */
    @MqttListener(topics = "confirmations/#")
    public void handleConfirmation(String topic, MessageWrapper wrapper) {
        MessageStore message = messageStoreRepository.findByMessageId(wrapper.getMessageId());
        if (message != null) {
            message.setStatus(MessageStatus.CONFIRMED);
            message.setConfirmedAt(System.currentTimeMillis());
            messageStoreRepository.save(message);
            log.info("消息已确认: {}", wrapper.getMessageId());
        }
    }
    
    private void scheduleConfirmationCheck(String messageId) {
        retryScheduler.schedule(() -> {
            MessageStore message = messageStoreRepository.findByMessageId(messageId);
            if (message != null && message.getStatus() == MessageStatus.PENDING) {
                // 重新发送消息
                log.warn("消息未确认，重新发送: {}", messageId);
                mqttTemplate.send(message.getTopic(), 
                    new MessageWrapper(message.getPayload(), messageId), 1, false);
                
                // 再次安排检查
                scheduleConfirmationCheck(messageId);
            }
        }, 30, TimeUnit.SECONDS);
    }
    
    @Data
    @AllArgsConstructor
    public static class MessageWrapper {
        private Object payload;
        private String messageId;
    }
    
    public enum MessageStatus {
        PENDING, CONFIRMED, FAILED
    }
}

/**
 * MQTT连接管理器
 * 支持多Broker负载均衡
 */
@Component
@Slf4j
public class MQTTConnectionManager {
    
    @Value("${mqtt.brokers}")
    private String[] brokerUrls;
    
    private final Map<String, MqttClient> clients = new ConcurrentHashMap<>();
    private final AtomicInteger currentIndex = new AtomicInteger(0);
    
    /**
     * 获取可用的MQTT客户端
     */
    public MqttClient getClient(String clientId) throws MqttException {
        return clients.computeIfAbsent(clientId, id -> {
            try {
                String brokerUrl = getNextBrokerUrl();
                MqttClient client = new MqttClient(brokerUrl, id, new MemoryPersistence());
                client.connect(createConnectOptions());
                
                log.info("创建MQTT客户端连接: {} -> {}", id, brokerUrl);
                return client;
            } catch (MqttException e) {
                log.error("创建MQTT客户端失败", e);
                throw new RuntimeException("连接创建失败", e);
            }
        });
    }
    
    /**
     * 轮询获取Broker URL
     */
    private String getNextBrokerUrl() {
        int index = currentIndex.getAndUpdate(i -> (i + 1) % brokerUrls.length);
        return brokerUrls[index];
    }
    
    /**
     * 创建连接选项
     */
    private MqttConnectOptions createConnectOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true);
        options.setAutomaticReconnect(true);
        options.setConnectionTimeout(30);
        options.setKeepAliveInterval(60);
        options.setMaxInflight(1000);
        return options;
    }
    
    /**
     * 关闭所有连接
     */
    @PreDestroy
    public void shutdown() {
        clients.forEach((id, client) -> {
            try {
                if (client.isConnected()) {
                    client.disconnect();
                }
                client.close();
                log.info("关闭MQTT客户端: {}", id);
            } catch (MqttException e) {
                log.warn("关闭客户端失败: {}", id, e);
            }
        });
        clients.clear();
    }
}

/**
 * MQTT SSL配置
 */
@Configuration
public class MQTTSslConfig {
    
    @Value("${mqtt.ssl.keystore}")
    private String keystorePath;
    
    @Value("${mqtt.ssl.keystore-password}")
    private String keystorePassword;
    
    @Value("${mqtt.ssl.truststore}")
    private String truststorePath;
    
    @Value("${mqtt.ssl.truststore-password}")
    private String truststorePassword;
    
    @Bean
    public SSLSocketFactory sslSocketFactory() throws Exception {
        // 加载密钥库
        KeyStore keyStore = KeyStore.getInstance("JKS");
        try (InputStream keyStoreStream = new FileInputStream(keystorePath)) {
            keyStore.load(keyStoreStream, keystorePassword.toCharArray());
        }
        
        // 加载信任库
        KeyStore trustStore = KeyStore.getInstance("JKS");
        try (InputStream trustStoreStream = new FileInputStream(truststorePath)) {
            trustStore.load(trustStoreStream, truststorePassword.toCharArray());
        }
        
        // 创建SSL上下文
        SSLContext sslContext = SSLContext.getInstance("TLS");
        TrustManagerFactory trustManagerFactory = 
            TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
        trustManagerFactory.init(trustStore);
        
        KeyManagerFactory keyManagerFactory = 
            KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
        keyManagerFactory.init(keyStore, keystorePassword.toCharArray());
        
        sslContext.init(keyManagerFactory.getKeyManagers(), 
                       trustManagerFactory.getTrustManagers(), 
                       new SecureRandom());
        
        return sslContext.getSocketFactory();
    }
    
    @Bean
    public MqttConnectOptions mqttConnectOptions(SSLSocketFactory sslSocketFactory) {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setSocketFactory(sslSocketFactory);
        options.setHttpsHostnameVerificationEnabled(false); // 仅测试环境
        return options;
    }
}

/**
 * MQTT认证拦截器
 */
@Component
public class MQTTAuthInterceptor implements MqttInterceptor {
    
    @Autowired
    private JwtTokenProvider tokenProvider;
    
    @Autowired
    private TopicAuthorizationService authService;
    
    @Override
    public boolean prePublish(MqttClient client, MqttPublishMessage message) {
        String username = getUsernameFromClient(client);
        String topic = message.getTopicName();
        
        if (!authService.canPublish(username, topic)) {
            log.warn("用户 {} 无权限发布到主题: {}", username, topic);
            return false;
        }
        
        return true;
    }
    
    @Override
    public boolean preSubscribe(MqttClient client, MqttSubscribeMessage message) {
        String username = getUsernameFromClient(client);
        
        for (MqttTopicSubscription subscription : message.topics()) {
            if (!authService.canSubscribe(username, subscription.topicName())) {
                log.warn("用户 {} 无权限订阅主题: {}", username, subscription.topicName());
                return false;
            }
        }
        
        return true;
    }
    
    private String getUsernameFromClient(MqttClient client) {
        // 从客户端连接信息中提取用户名
        return client.getClientOptions().getUserName();
    }
}

/**
 * 主题授权服务
 */
@Service
@Slf4j
public class TopicAuthorizationService {
    
    private final Map<String, Set<String>> userPublishPermissions = new ConcurrentHashMap<>();
    private final Map<String, Set<String>> userSubscribePermissions = new ConcurrentHashMap<>();
    
    /**
     * 检查发布权限
     */
    public boolean canPublish(String username, String topic) {
        Set<String> allowedTopics = userPublishPermissions.get(username);
        return allowedTopics != null && matchesPattern(allowedTopics, topic);
    }
    
    /**
     * 检查订阅权限
     */
    public boolean canSubscribe(String username, String topic) {
        Set<String> allowedTopics = userSubscribePermissions.get(username);
        return allowedTopics != null && matchesPattern(allowedTopics, topic);
    }
    
    /**
     * 添加用户权限
     */
    public void addUserPermissions(String username, 
                                 Set<String> publishTopics, 
                                 Set<String> subscribeTopics) {
        userPublishPermissions.put(username, publishTopics);
        userSubscribePermissions.put(username, subscribeTopics);
        log.info("为用户 {} 添加权限: 发布={}, 订阅={}", 
                username, publishTopics, subscribeTopics);
    }
    
    private boolean matchesPattern(Set<String> patterns, String topic) {
        return patterns.stream().anyMatch(pattern -> 
            topic.matches(pattern.replace("+", "[^/]+").replace("#", ".+")));
    }
}

/**
 * MQTT性能调优配置
 */
@Configuration
public class MQTTPerformanceConfig {
    
    @Bean
    public MqttConnectOptions highPerformanceOptions(
            @Value("${mqtt.performance.max-inflight:1000}") int maxInflight,
            @Value("${mqtt.performance.buffer-size:16384}") int bufferSize) {
        
        MqttConnectOptions options = new MqttConnectOptions();
        options.setMaxInflight(maxInflight);
        options.setAutomaticReconnect(true);
        options.setConnectionTimeout(10);
        options.setKeepAliveInterval(30);
        options.setCleanSession(true);
        
        // 高性能配置
        options.setExecutorServiceTimeout(1000);
        return options;
    }
    
    @Bean
    public ThreadPoolTaskExecutor mqttTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(20);
        executor.setMaxPoolSize(100);
        executor.setQueueCapacity(1000);
        executor.setThreadNamePrefix("mqtt-");
        executor.initialize();
        return executor;
    }
    
    @Bean
    public MqttClientFactory mqttClientFactory(MqttConnectOptions options) {
        DefaultMqttClientFactory factory = new DefaultMqttClientFactory();
        factory.setConnectionOptions(options);
        
        // 设置自定义线程池
        factory.setTaskExecutor(mqttTaskExecutor());
        
        return factory;
    }
}

/**
 * MQTT监控指标收集器
 */
@Component
@Slf4j
public class MQTTMetricsCollector {
    
    private final MeterRegistry meterRegistry;
    private final Map<String, Timer.Sample> messageTimers = new ConcurrentHashMap<>();
    
    public MQTTMetricsCollector(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
        initMetrics();
    }
    
    private void initMetrics() {
        // 连接数指标
        Gauge.builder("mqtt.connections.active", () -> 
                MQTTConnectionManager.getActiveConnectionCount())
             .description("活跃MQTT连接数")
             .register(meterRegistry);
        
        // 消息速率指标
        Meter.builder("mqtt.messages.published")
             .description("发布的消息数量")
             .register(meterRegistry);
        
        Meter.builder("mqtt.messages.received")
             .description("接收的消息数量")
             .register(meterRegistry);
    }
    
    /**
     * 记录消息开始处理
     */
    public void recordMessageStart(String messageId) {
        messageTimers.put(messageId, Timer.start(meterRegistry));
    }
    
    /**
     * 记录消息处理完成
     */
    public void recordMessageEnd(String messageId, String topic, boolean success) {
        Timer.Sample sample = messageTimers.remove(messageId);
        if (sample != null) {
            sample.stop(Timer.builder("mqtt.message.processing.time")
                .tag("topic", topic)
                .tag("success", Boolean.toString(success))
                .register(meterRegistry));
        }
    }
    
    /**
     * 记录连接事件
     */
    public void recordConnectionEvent(String clientId, boolean connected) {
        Counter.builder("mqtt.connection.events")
            .tag("clientId", clientId)
            .tag("status", connected ? "connected" : "disconnected")
            .register(meterRegistry)
            .increment();
        
        log.info("客户端 {} {}", clientId, connected ? "连接" : "断开连接");
    }
}

/**
 * MQTT健康检查服务
 */
@Service
@Slf4j
public class MQTTHealthService {
    
    @Autowired
    private MqttClientFactory clientFactory;
    
    private final ScheduledExecutorService healthChecker = 
        Executors.newSingleThreadScheduledExecutor();
    
    private volatile boolean isHealthy = false;
    private final AtomicInteger consecutiveFailures = new AtomicInteger(0);
    
    @PostConstruct
    public void startHealthCheck() {
        healthChecker.scheduleAtFixedRate(this::checkHealth, 0, 30, TimeUnit.SECONDS);
    }
    
    private void checkHealth() {
        try (MqttClient client = clientFactory.createClient()) {
            MqttConnectOptions options = clientFactory.getConnectionOptions();
            client.connect(options);
            
            // 测试消息发布
            client.publish("healthcheck", new MqttMessage("test".getBytes()));
            
            isHealthy = true;
            consecutiveFailures.set(0);
            log.debug("MQTT健康检查通过");
            
        } catch (Exception e) {
            int failures = consecutiveFailures.incrementAndGet();
            isHealthy = false;
            
            log.warn("MQTT健康检查失败 ({}): {}", failures, e.getMessage());
            
            if (failures >= 3) {
                log.error("MQTT服务连续失败，触发故障转移");
                triggerFailover();
            }
        }
    }
    
    private void triggerFailover() {
        // 实现故障转移逻辑
        log.info("执行MQTT故障转移");
        // 1. 切换到备用Broker
        // 2. 通知监控系统
        // 3. 记录故障事件
    }
    
    public boolean isHealthy() {
        return isHealthy;
    }
    
    @PreDestroy
    public void shutdown() {
        healthChecker.shutdown();
        try {
            if (!healthChecker.awaitTermination(10, TimeUnit.SECONDS)) {
                healthChecker.shutdownNow();
            }
        } catch (InterruptedException e) {
            healthChecker.shutdownNow();
        }
    }
}