Rhythmix（流式数据规则表达式），一行就够了！

Rhythmix（流式数据规则表达式），一行就够了！

视频概览讲解：www.bilibili.com/video/BV1S3…

详细文档（有疑问可以看看这个）：github.com/MFinnnne/rh…

什么是 Rhythmix？

在实际开发中，我们经常会遇到这样的需求：

• ️ 温度监控：当温度连续 3 次超过 30°C 时触发告警
• 生产线质量控制：检测产品重量是否在 [95,105] 克范围内
• 网络监控：API 响应时间连续 5 次超过 1000ms 时发送通知
• 金融风控：检测异常交易模式，如金额突然从低到高的状态转换

让我们看看这个复杂的状态机需求：某个数值依次要满足 先大于1 → 然后累计3次小于1 → 最后等于3

传统的做法是什么？写一堆 if-else，维护各种状态变量，代码又臭又长。

// 传统写法：需要维护多个状态变量和复杂的状态机逻辑
enum State { INIT, HIGH_LOAD, COUNTING_LOW, STABLE }
State currentState = State.INIT;
int lowLoadCount = 0;

for (Data data : dataList) {
    double value = data.getValue();

    switch (currentState) {
        case INIT:
            if (value > 1) {
                currentState = State.HIGH_LOAD;
            }
            break;
        case HIGH_LOAD:
            if (value < 1) {
                currentState = State.COUNTING_LOW;
                lowLoadCount = 1;
            }
            break;
        case COUNTING_LOW:
            if (value < 1) {
                lowLoadCount++;
                if (lowLoadCount >= 3) {
                    currentState = State.STABLE;
                }
            } else {
                // 中断了，回到高负载状态
                currentState = State.HIGH_LOAD;
                lowLoadCount = 0;
            }
            break;
        case STABLE:
            if (value == 3) {
                // 检测成功！
                return true;
            }
            break;
    }
}

// Rhythmix 写法：一行表达式搞定复杂状态机！
String expression = "{>1}->{count(<1,3)}->{==3}";
RhythmixExecutor executor = RhythmixCompiler.compile(expression);
boolean result = executor.execute(dataList);

表达式解读：

• {>1} → 第一个状态单元：值大于 1（高负载）

• {count(<1,3)} → 第二个状态单元：累计 3 次值小于 1（低负载计数）

• {==3} → 第三个状态单元：值等于 3（稳定状态）

  看到了吗？40+ 行的复杂状态机代码，用 Rhythmix 一行表达式就搞定了！ 这就是 Rhythmix 的魅力所在！下面的动图展示了表达式的运行过程：

什么是"状态单元"？

在 Rhythmix 中，状态单元（State Unit） 是表达式的基本组成部分。每个用 {} 包裹的条件就是一个状态单元。

Rhythmix 的灵活性在于：你既可以使用完整的状态转换表达式，也可以使用单个状态单元！

// 方式一：使用单个状态单元（简单场景）
String expression1 = ">30";                                   // 基础比较状态单元
String expression2 = "count!(>30,3)";                         // 计数状态单元
String expression3 = "[95,105]";                              // 区间状态单元
String expression4 = "filter(>0).window(5).avg().meet(>50)"; // 链式表达式状态单元

// 方式二：使用状态转换表达式（组合多个状态单元）
String expression5 = "{>1}->{count(<1,3)}->{==3}";           // 三个状态单元的转换
String expression6 = "{<=40}->{>80}";                         // 两个状态单元的转换
String expression7 = "{[95,105]}->{(0,95)||(105,200)}";     // 组合逻辑运算的状态单元
String expression8 = "{>0}->{filter(>5).window(3).avg().meet(>10)}"; // 组合链式表达式状态单元

状态单元的类型：

• 基础比较：>30、<10、==5，<=,>=
• 区间表达式：[95,105]、(20,30)
• 逻辑运算：<10||>40、>=20&&<=30
• 状态函数：count(>4,3)、count!(>4,3)
• 链式表达式：filter().window().avg().meet()

选择建议：

• 简单判断 → 使用单个状态单元（如 >30）
• 状态转换 → 组合多个状态单元（如 {>1}->{<1}）
• 复杂数据处理 → 使用链式表达式状态单元（如 filter().window().avg()）

快速上手

1. 添加依赖

在你的 pom.xml 中添加：

<dependency>
    <groupId>io.github.mfinnnne</groupId>
    <artifactId>rhythmix</artifactId>
    <version>1.0.1</version>
</dependency>

2. 三步走，轻松使用

// 第一步：编译表达式
String expression = "count(>4,3)";
RhythmixExecutor executor = RhythmixCompiler.compile(expression);

// 第二步：创建事件数据
RhythmixEventData data1 = new RhythmixEventData("event1", "5", new Timestamp(System.currentTimeMillis()));
RhythmixEventData data2 = new RhythmixEventData("event2", "6", new Timestamp(System.currentTimeMillis()));
RhythmixEventData data3 = new RhythmixEventData("event3", "7", new Timestamp(System.currentTimeMillis()));

// 第三步：执行表达式
boolean result = executor.execute(data1); // 返回 false
boolean result = executor.execute(data2); // 返回 false
boolean result = executor.execute(data3); // 返回 true

就这么简单！三步搞定一个规则判断。

核心功能详解

1. 基础比较运算符（最简单的状态单元）

支持所有常见的比较操作：

">30"           // 大于 30
"<10"           // 小于 10
">=100"         // 大于等于 100
"<=50"          // 小于等于 50
"==25"          // 等于 25
"!=0"           // 不等于 0

实际应用场景：

• 温度监控：>30 检测高温告警
• 库存管理：<10 检测库存不足
• 错误过滤：!=0 排除错误值

2. 区间表达式

支持四种区间类型：

"(20,25)"       // 开区间：20 < x < 25
"(20,25]"       // 左开右闭：20 < x ≤ 25
"[20,25)"       // 左闭右开：20 ≤ x < 25
"[20,25]"       // 闭区间：20 ≤ x ≤ 25

实际应用场景：

• 质量控制：[95,105] 产品重量规格检测
• 网络延迟：[0,500) 正常响应时间范围
• 温度控制：(20,30) 舒适温度区间

3. 逻辑运算符

组合多个条件，实现复杂逻辑：

"<10||>40"                  // 或：太冷或太热
">=20&&<=30"                // 与：正常范围
"((1,7]||>10)&&!=5"         // 复杂组合

实际应用场景：

• 异常检测：<10||>40 温度异常（过冷或过热）
• 正常范围：>=20&&<=30 温度在正常范围内
• 排除错误码：!=0&&!=(-1) 排除多个错误值

4. 状态单元函数

count() - 非连续计数

计数器会持续累加，即使条件不满足也不会重置：

"count(>4,3)"   // 统计 3 次大于 4 的值（可以不连续）

行为示例：

数据流：5, 2, 6, 1, 7
计数：  1, 1, 2, 2, 3   （第 5 个数据时返回 true）

count!() - 连续计数

计数器在条件不满足时会重置为 0：

"count!(>4,3)"  // 统计连续 3 次大于 4 的值

行为示例：

数据流：5, 2, 6, 7, 8
计数：  1, 0, 1, 2, 3   （需要连续 3 次才返回 true）

5. 链式表达式（特殊的状态单元）

链式表达式本身也是一种状态单元！它可以构建复杂的数据处理管道：

"filter(>0).window(3).avg().meet(>50)"
// 过滤正值 → 取最近 3 个 → 计算平均值 → 检查是否大于 50

链式组件详解：

• filter(): 数据过滤

  "filter(>0)"              // 保留正值
  "filter([0,100])"         // 保留范围内的值
  

• limit(): 队列管理

  "limit(10)"               // 保留最近 10 个值
  "limit(100ms)"            // 保留最近 100 毫秒的数据
  "limit(5s)"               // 保留最近 5 秒的数据
  

• window(): 滑动窗口

  "window(5)"               // 5 个值的滑动窗口
  "window(100ms)"           // 100 毫秒的时间窗口
  

• 计算器: 聚合计算

  "sum()"                   // 求和
  "avg()"                   // 平均值
  "count()"                 // 计数
  "stddev()"                // 标准差
  

• meet(): 条件检查

  "meet(>10)"               // 结果大于 10
  "meet([5,50])"            // 结果在范围内
  

实际应用场景：

// 温度监控：最近 5 个正值的平均温度是否超过 50°C
"filter(>0).window(5).avg().meet(>50)"

// 质量控制：最近 20 个产品中，取最近 10 个的平均重量是否在 [98,102] 范围内
"filter([95,105]).limit(20).window(10).avg().meet([98,102])"

// 网络监控：最近 100ms 内正值的总和是否大于等于 7
"filter(>0).window(100ms).sum().meet(>=7)"

自定义函数（扩展链式表达式）

Rhythmix 支持自定义函数来扩展链式表达式的能力，让你可以实现特定业务逻辑！

三种自定义函数类型

1. 自定义过滤器（FilterUDF）

实现自己的数据过滤逻辑：

public class TemperatureFilterUDF implements ChainFilterUDF {
    @Override
    public String getName() {
        return "tempFilter"; // 函数名称
    }

    @Override
    public boolean filter(RhythmixEventData event) {
        double temp = Double.parseDouble(event.getValue());
        return temp >= 20.0 && temp <= 80.0; // 保留 20-80 度的数据
    }
}

// 在表达式中使用
"tempFilter().sum().meet(>100)"

2. 自定义计算器（CalculatorUDF）

实现自己的数据计算逻辑：

public class MaxCalculator implements ChainCalculatorUDF {
    @Override
    public String getName() {
        return "myMax"; // 函数名称
    }

    @Override
    public Number calculate(List<RhythmixEventData> values) {
        // 找出最大值
        return values.stream()
            .mapToDouble(v -> Double.parseDouble(v.getValue()))
            .max()
            .orElse(0);
    }
}

// 在表达式中使用
"filter(>0).myMax().meet(>100)"

3. 自定义条件判断（MeetUDF）

实现自己的条件判断逻辑：

public class CustomThresholdMeetUDF implements ChainMeetUDF {
    @Override
    public String getName() {
        return "customMeet"; // 函数名称
    }

    @Override
    public boolean meet(Number calculatedValue) {
        double value = calculatedValue.doubleValue();
        // 自定义判断：值必须是偶数且大于 10
        return value > 10 && value % 2 == 0;
    }
}

// 在表达式中使用
"filter(>0).sum().customMeet()"

使用说明：

• 自定义函数会被系统自动发现和注册，无需手动配置
• 函数名称必须唯一
• 可以在链式表达式的任意位置使用自定义函数
• 支持与内置函数混合使用

实际应用示例：

// 组合使用：自定义过滤 + 内置计算 + 自定义判断
"tempFilter().window(5).avg().customMeet()"

// 复杂业务逻辑：多个自定义函数组合
"tempFilter().limit(10).myMax().customMeet()"

状态单元的使用方式

前面介绍了 5 种类型的状态单元，现在来看看如何使用它们：

方式一：单独使用状态单元

最简单的方式，直接使用一个状态单元进行判断：

// 基础比较
String expression1 = ">30";

// 区间判断
String expression2 = "[95,105]";

// 计数判断
String expression3 = "count!(>30,3)";

// 链式表达式
String expression4 = "filter(>0).window(5).avg().meet(>50)";

方式二：状态转换（组合多个状态单元）

这是 Rhythmix 最强大的功能！通过 -> 连接多个状态单元，构建复杂的状态机。

语法格式：{状态单元1}->{状态单元2}->{状态单元3}...

// 两个状态的转换
"{<=40}->{>80}"                     // 从正常温度到过热

// 三个状态的转换
"{>1}->{count(<1,3)}->{==3}"       // 高负载 → 3次低负载 → 稳定

// 复杂状态单元的组合
"{==0}->{count!(>4,3)}"             // 从空闲到活跃（连续3次高值）
"{[95,105]}->{(0,95)||(105,200)}"  // 从合格到不合格（带逻辑运算）

// 链式表达式也可以作为状态单元
"{>0}->{filter(>5).window(3).avg().meet(>10)}"

核心概念：

• 每个 {} 包裹的条件都是一个状态单元
• 使用 -> 连接状态单元，表示状态转换
• 只有当前状态满足后，才会进入下一个状态
• 最后一个状态满足时，整个表达式返回 true

实际应用场景：

• 设备监控：检测设备从正常状态突然进入异常状态
• 生产线：检测从空闲状态进入生产状态
• 质量控制：检测产品质量从合格变为不合格
• 用户行为：检测用户从浏览到购买的转化路径

真实场景应用

我在项目中准备了 15+ 个真实场景的测试用例，涵盖了各行各业：

1. 智能家居 - 温度控制

// 场景：温度连续 3 次超过 26°C 时开启空调
String expression = "count!(>26,3)";

2. 制造业 - 生产线质量控制

// 场景：产品重量从合格范围突然变为不合格
String expression = "{[95,105]}->{(0,95)||(105,200)}";

3. 网络监控 - API 响应时间

// 场景：检测 2 次慢响应（>1000ms）
String expression = "count(>1000,2)";

4. 医疗健康 - 心率监控

// 场景：心率从正常范围突然进入异常范围
String expression = "{[60,100]}->{<50||>120}";

5. 金融风控 - 欺诈检测

// 场景：交易金额从小额突然变为大额
String expression = "{<100}->{>10000}";

6. 环境监测 - 空气质量

// 场景：PM2.5 连续 5 次超标（>75）
String expression = "count!(>75,5)";

7. 数据中心 - CPU 监控

// 场景：最近 5 个 CPU 使用率的平均值超过 80%
String expression = "filter(>0).window(5).avg().meet(>80)";

性能与优势

为什么选择 Rhythmix？

1. 简洁优雅：一行表达式替代几十行代码
2. 易于维护：规则变更只需修改表达式字符串
3. 功能强大：支持复杂的状态转换和链式处理
4. 零依赖：轻量级，不引入额外依赖

对比传统方案

最佳实践

1. 选择合适的表达式类型

• 简单阈值检测 → 使用基础比较运算符
• 范围验证 → 使用区间表达式
• 模式检测 → 使用计数函数
• 复杂聚合 → 使用链式表达式

2. 性能优化建议

• 使用 limit() 防止内存占用过大
• 使用 window() 进行滑动窗口计算
• 在链式表达式中尽早使用 filter() 过滤数据

3. 测试建议

• 测试边界值（临界情况）
• 测试状态转换
• 使用真实数据模式进行测试

项目资源

• GitHub 仓库: github.com/MFinnnne/rh…
• 完整示例项目: 包含 68+ 个测试用例，涵盖所有功能

未来计划

Rhythmix 还在持续进化中！以下是我们计划中的新特性，欢迎大家提出建议和反馈！

1. 状态单元函数自定义

目标：让用户可以自定义状态单元级别的函数，而不仅仅是链式表达式中的函数。

应用场景：

// 自定义状态单元函数
"customStateFunc(>10,3)" // 类似 count，但有自己的逻辑

// 在状态转换中使用
"{customStateFunc(>10,3)}->{<5}"

预期收益：

• 更灵活的状态检测逻辑
• 可复用的业务规则封装
• 更强的表达能力

2. 多事件源支持

目标：支持同时监控多个数据源，实现跨传感器的复杂规则判断。

语法设计（征求意见中）：

方案一：使用花括号嵌套

// a 和 b 是不同传感器的别名
{{a:==0}&&{b:==1}}

// 实际应用：温度传感器正常且湿度传感器异常
{{temp:<30}&&{humidity:>80}}

方案二：使用竖线分隔

// 使用 | 作为事件源标识符
{|a:==0|&&|b:==1|}

// 实际应用
{|temp:<30|&&|humidity:>80|}

你更喜欢哪种语法？欢迎在评论区留言！

应用场景：

• 工业监控：同时监控温度、压力、振动多个传感器
• 智能家居：联动多个设备的状态（门窗、烟雾、温度）
• 网络监控：关联多个服务的健康状态
• 金融风控：综合多个指标判断风险

示例：

// 当温度传感器超标且压力传感器异常时触发告警
{{temp:>80}&&{pressure:>150}}

// 门窗打开且烟雾检测到异常
{{door:==1}&&{smoke:>50}}

// 三个传感器的复杂组合
{{a:>10}&&{b:<5}||{c:[20,30]}}

3. ️ 高可用支持

目标：确保系统在故障恢复后能够继续正确工作，不丢失状态。

实现方案：

• 消息队列集成：接入 Kafka、RabbitMQ 等消息队列
• 状态持久化：使用 Redis 或数据库保存表达式执行状态
• 故障恢复：系统重启后自动恢复到故障前的状态

应用场景：

// 生产环境中的关键监控
// 即使系统重启，也能继续追踪"连续3次异常"的状态
count!(>80, 3)

// 复杂状态转换不会因为宕机而重置
{[20,30]}->{count!(>50,5)}->{<10}

预期收益：

• 生产环境可靠性大幅提升
• 支持分布式部署
• 数据不丢失，状态可恢复

4. 其他规划中的特性

• 性能监控面板：实时查看表达式执行情况
• 更多内置函数：中位数、百分位数、移动平均等

你的想法很重要！

如果你有任何建议或想法，欢迎：

• 在评论区留言讨论
• 在 GitHub 提交 Issue
• 提交 Pull Request 参与贡献

让我们一起让 Rhythmix 变得更强大！

总结

Rhythmix 是一个专为流式数据处理设计的规则表达式引擎，它让复杂的事件处理变得简单而优雅。无论你是在做 IoT 数据处理、实时监控、异常检测，还是质量控制，Rhythmix 都能帮你大幅简化代码，提高开发效率。

如果你觉得这个项目对你有帮助，欢迎：

• ⭐ 给项目点个 Star
• 提交 Issue 和 PR
• 分享给更多的开发者

最后，附上项目地址：github.com/MFinnnne/rh…

让我们一起用 Rhythmix 让代码更优雅！

我不知道我开源这个玩意可以干嘛，思考再三还是开源了出来，或许只是自我感动吧，毕竟工作中你写了这些也不会有人买单，但是自我感动也算是感动吧。愿各位猿真的能心有花朵吧