物联网实战|Spring Boot+Netty搭建MQTT核心服务层

{
  "sub": "{productKey}/{deviceKey}",  // 设备唯一标识
  "productKey": "product-001",
  "deviceKey": "device-001",
  "iat": 1750051200,                  // 签发时间
  "exp": 1750137600,                  // 过期时间（1天后）
  "scope": ["publish", "subscribe"]   // 权限范围
}

public interface Authenticator {
    /**
     * 认证设备
     * @return 认证成功返回设备信息，否则返回null
     */
    DeviceInfo authenticate(String productKey, String deviceKey, String password);
    
    /**
     * 验证Token
     */
    boolean validateToken(String token);
}

// 设备在线状态缓存
ConcurrentHashMap<String, Boolean> deviceOnlineCache;  // key: "productKey:deviceKey"

// 在线时间记录
ConcurrentHashMap<String, Long> deviceOnlineTimeCache;  // key: "productKey:deviceKey"

public interface DeviceShadowManager {
    void initializeShadow(String productKey, String deviceKey);
    Map<String, Object> getShadow(String productKey, String deviceKey);
    void updateDesired(String productKey, String deviceKey, Map<String, Object> desired);
    void updateReported(String productKey, String deviceKey, Map<String, Object> reported);
    Map<String, Object> syncShadow(String productKey, String deviceKey);
    void deleteShadow(String productKey, String deviceKey);
}

public interface RuleEngine {
    /**
     * 初始化规则引擎
     */
    void initialize();
    
    /**
     * 评估规则
     * @param topic 消息主题
     * @param payload 消息内容
     * @return 匹配结果
     */
    EvaluationResult evaluate(String topic, byte[] payload);
    
    /**
     * 添加规则
     */
    void addRule(Rule rule);
    
    /**
     * 移除规则
     */
    void removeRule(String ruleId);
}

@Data
public class Rule {
    private String ruleId;           // 规则ID
    private String ruleName;         // 规则名称
    private String ruleType;         // 规则类型
    private String status;           // 状态：ENABLED/DISABLED
    private Condition condition;     // 触发条件
    private List<Action> actions;    // 执行动作列表
    
    @Data
    public static class Condition {
        private String topic;              // 监听主题
        private String expression;         // 条件表达式
        private Map<String, String> params; // 参数
    }
    
    @Data
    public static class Action {
        private String actionType;   // 动作类型：PUBLISH/SAVE_DB/SMS/EMAIL
        private String target;       // 目标（主题/服务地址等）
        private Map<String, Object> params; // 动作参数
    }
}

public interface TimeSeriesWriter {
    /**
     * 写入时序数据
     * @param metric 度量名称
     * @param tags 标签（维度）
     * @param fields 字段（指标值）
     * @param timestamp 时间戳
     */
    void write(String metric, Map<String, String> tags, Map<String, Object> fields, long timestamp);
}

@Component
public class DeviceStatusEventListener {
    @EventListener
    public void onDeviceOnline(DeviceOnlineEvent event) {
        // 设备上线时初始化影子
        shadowService.initializeShadow(event.getProductKey(), event.getDeviceKey());
    }
}

@Component
public class RuleEventListener {
    @EventListener
    public void onMessage(MessageEvent event) {
        // 触发规则评估
        ruleEngine.evaluate(event.getTopic(), event.getPayload());
    }
}

@Configuration
public class AsyncConfig implements AsyncConfigurer {
    @Override
    public Executor getAsyncExecutor() {
        TaskExecutor executor = new TaskExecutor();
        executor.setTaskDecorator(new TraceIdTaskDecorator());
        return executor;
    }
}

@Component
public class LdapAuthenticator implements Authenticator {
    @Override
    public DeviceInfo authenticate(String productKey, String deviceKey, String password) {
        // LDAP 认证逻辑
    }
    
    @Override
    public boolean validateToken(String token) {
        // Token 验证逻辑
    }
}

@ActionHandler("CUSTOM_ACTION")
public void handleCustomAction(Action action, EvaluationResult result) {
    // 自定义动作处理逻辑
}

@Component
@ConditionalOnProperty(name = "storage.mode", havingValue = "mysql")
public class MySqlTimeSeriesWriter implements TimeSeriesWriter {
    // MySQL 实现
}