@Component
public class MqttBrokerServer {
    private static final int PORT = 1883;

    @PostConstruct
    public void start() throws InterruptedException {
        // 加载规则
        sqlRuleEngine.loadRules();
        
        // 启动 Netty 服务
        new ServerBootstrap()
            .group(bossGroup, workerGroup)
            .channel(NioServerSocketChannel.class)
            .childHandler(...)
            .bind(PORT)
            .sync();
    }
}

private void handlePublish(ChannelHandlerContext ctx, MqttPublishMessage msg) {
    String topic = msg.variableHeader().topicName();
    String payloadStr = new String(payload, StandardCharsets.UTF_8);
    String clientId = clientIdMap.getOrDefault(ctx.channel(), "unknown");

    // 1. 规则引擎处理
    sqlRuleEngine.process(topic, clientId, payloadStr);
    // 2. 消息转发
    subscriptionManager.publishToSubscribers(topic, payload, packetId);
}

public void publishToSubscribers(String topic, byte[] payload, int packetId) {
    for (Map.Entry<String, Set<Channel>> entry : subscriptions.entrySet()) {
        String filter = entry.getKey();
        if (matchTopic(topic, filter)) { // 主题匹配
            // 转发消息给订阅者
            for (Channel ch : entry.getValue()) {
                ch.writeAndFlush(publishMsg.retain());
            }
        }
    }
}

@Service
public class SqlRuleEngine {
    // 从数据库加载规则
    public void loadRules() {
        List<RuleConfig> configs = ruleConfigRepository.findAllEnabledRules();
        List<SqlRule> rules = configs.stream().map(this::toSqlRule).collect(Collectors.toList());
        ruleCache.loadRules(rules);
    }

    // 处理消息
    public void process(String topic, String clientId, String payload) {
        // 规则匹配与执行
        List<RuleMatchResult> results = ruleExecutor.execute(topic, payload);
        // 动作调度
        actionDispatcher.dispatch(results, clientId);
    }
}

public class RuleCache {
    private final Map<Long, SqlRule> rules = new ConcurrentHashMap<>();
    private final Map<Long, RuleParseResult> parseResults = new ConcurrentHashMap<>();

    public void loadRules(List<SqlRule> sqlRules) {
        for (SqlRule rule : sqlRules) {
            if (rule.getEnabled() == 1) {
                rules.put(rule.getId(), rule);
                // 预解析条件表达式
                if (rule.getConditionExpr() != null && !rule.getConditionExpr().isEmpty()) {
                    Expression whereExpr = SqlRuleParser.parseCondition(rule.getConditionExpr());
                    parseResults.put(rule.getId(), RuleParseResult.builder().whereExpr(whereExpr).build());
                }
            }
        }
    }
}

public List<RuleMatchResult> execute(String topic, String payload) {
    List<RuleMatchResult> results = new ArrayList<>();
    for (SqlRule rule : ruleCache.getRules().values()) {
        RuleMatchResult result = executeSingleRule(rule, topic, payload);
        if (result.isMatched()) {
            results.add(result);
        }
    }
    return results;
}

private RuleMatchResult executeSingleRule(SqlRule rule, String topic, String payload) {
    // 1. 主题匹配 (FROM)
    if (!TopicMatcher.match(topic, rule.getTopicPattern())) {
        return RuleMatchResult.builder().matched(false).build();
    }
    
    // 2. 条件评估 (WHERE)
    boolean conditionMatched = ConditionEvaluator.evaluate(jsonMsg, parseResult.getWhereExpr());
    
    // 3. 字段提取 (SELECT)
    Map<String, Object> extractedFields = ConditionEvaluator.extractFields(jsonMsg, selectFields);
    
    return RuleMatchResult.builder()
        .matched(conditionMatched)
        .extractedFields(extractedFields)
        .build();
}

public static RuleParseResult parse(String sql) {
    Select select = (Select) CCJSqlParserUtil.parse(sql);
    PlainSelect plainSelect = (PlainSelect) select.getSelectBody();
    
    String fromTopic = plainSelect.getFromItem().toString();
    Expression whereExpr = plainSelect.getWhere();
    List<String> selectFields = extractSelectFields(plainSelect);
    
    return RuleParseResult.builder()
        .fromTopic(fromTopic)
        .whereExpr(whereExpr)
        .selectFields(selectFields)
        .build();
}

public static Expression parseCondition(String conditionExpr) {
    return CCJSqlParserUtil.parseCondExpression(conditionExpr);
}

public static boolean match(String msgTopic, String topicPattern) {
    String[] msgLevels = msgTopic.split("/");
    String[] filterLevels = topicPattern.split("/");
    
    for (int i = 0; i < filterLevels.length; i++) {
        if ("#".equals(filterLevels[i])) {
            return true; // # 匹配剩余所有
        } else if ("+".equals(filterLevels[i])) {
            continue; // + 匹配当前层级
        } else if (!filterLevels[i].equals(msgLevels[i])) {
            return false;
        }
    }
    return msgLevels.length == filterLevels.length;
}

public static boolean evaluate(JsonNode jsonMsg, Expression whereExpr) {
    if (whereExpr instanceof GreaterThan) {
        return evaluateGreaterThan(jsonMsg, (GreaterThan) whereExpr);
    } else if (whereExpr instanceof MinorThan) {
        return evaluateMinorThan(jsonMsg, (MinorThan) whereExpr);
    } else if (whereExpr instanceof EqualsTo) {
        return evaluateEqualsTo(jsonMsg, (EqualsTo) whereExpr);
    } else if (whereExpr instanceof AndExpression) {
        return evaluate(left) && evaluate(right);
    }
    return false;
}

public void dispatch(List<RuleMatchResult> results, String clientId) {
    for (RuleMatchResult result : results) {
        if (result.isMatched()) {
            executeActions(result, clientId);
        }
    }
}

private void executeActions(RuleMatchResult result, String clientId) {
    // 动作1: 保存消息记录
    saveDeviceData(result, clientId);
    // 动作2: 转发消息
    publishMessage(result);
}