CEP 客户端

CEPClient 是连接到实时规则引擎的 Java 库。它可以部署 EQL 规则、推送事件，并在规则命中时在你的进程内接收回调。它不轮询 REST。

项目	值
坐标	plantpulse-platform:plantpulse-cep-client
服务器	plantpulse-cep — 默认端口 7401 (TLS)
要求	Java 21
认证	X-API-Key 头部
线程	线程安全 · AutoCloseable

有两条通信路径

规则部署和查询走 HTTPS，事件发送和接收则通过与引擎共享的内存总线 (Valkey/Redis) 进行。因此，必须同时挂载 InMemoryClient 才能收到匹配回调。

开始使用

import java.util.Properties;
import plantpulse.cep.client.CEPClient;
import plantpulse.cep.client.eql.EQL;
import plantpulse.cep.client.eql.EQLUpdateListener;
import plantpulse.inmemory.InMemoryClient;
import plantpulse.inmemory.redis.RedisInMemoryClient;
import plantpulse.json.JSONObject;

Properties redis = new Properties();
redis.setProperty("cache.host", "plant-server.example.com");
redis.setProperty("cache.port", "6379");
redis.setProperty("cache.password", System.getenv("PP_CACHE_PASSWORD"));

RedisInMemoryClient bus = RedisInMemoryClient.builder().fromProperties(redis).buildClient();
bus.connect();

try (CEPClient client = CEPClient.builder()
        .baseUrl("https://plant-server.example.com:7401")
        .apiKey(System.getenv("PP_CEP_API_KEY"))
        .inmemory(bus)                       // build() 가 연결까지 합니다
        .build()) {

    EQL eql = new EQL("HIGH_TEMPERATURE",
            "SELECT * FROM Point WHERE tag_id = 'TAG_001' AND value > 80");
    EQLUpdateListener listener =
            (name, matches) -> System.out.println(name + " → " + matches);

    JSONObject deployed = client.deploy(eql, listener);
    if (!deployed.getBoolean("success")) {
        System.err.println("배포 실패: " + deployed.optString("message"));
        return;
    }

    JSONObject point = new JSONObject();
    point.put("tag_id", "TAG_001");
    point.put("timestamp", System.currentTimeMillis());
    point.put("value", "97.4");
    client.sendEvent("Point", point);

    Thread.sleep(1_000L);
    client.undeploy("HIGH_TEMPERATURE");
} finally {
    // Redisson 이 논데몬 스레드를 붙잡습니다 — 반드시 명시적으로 내립니다.
    bus.shutdown();
}

如果遗漏 shutdown()，进程将不会结束

内存总线会启动非守护线程。即使关闭 CEPClient，该线程仍会保留，导致 JVM 无法退出。这在批处理任务中尤其明显。

能做什么

方法	功能
deploy(EQL, listener)	规则部署 + 匹配回调注册
validate(EQL)	仅在服务器上编译 — 不部署
undeploy(name)	规则解除 + 监听器移除
isDeployed(name) · getEQL(name) · getText(name)	状态查询
getEQLMap() · getEQLMapCount()	所有已部署的规则
sendEvent(type, json)	事件发送 (单条 · 批量)
setApiKey(key)	运行时密钥轮换

如果密钥为 null 或为空，则不附加头部 — 也与关闭了认证的服务器兼容。

批量传输调优

sendEvent 内部会进行批量发送。当每秒推入数万条时进行调优。

参数	默认值	含义
shardCount	4	flush 工作线程数
queueSize	600,000	总队列大小
flushTriggerSize	8,000	达到此大小时触发 flush
maxInflightBatches	64	可同时进行的 flush 数量
batchIntervalMs	50	基于时间的 flush 间隔
skipResult	true	不等待结果 (fire-and-forget)

CEPClient client = CEPClient.builder()
        .host("plant-server.example.com")
        .apiKey(System.getenv("PP_CEP_API_KEY"))
        .inmemory(bus)
        .shardCount(8)
        .flushBatchSize(16_000)
        .queueSize(1_000_000)
        .build();

队列满了会怎样

超过 queueSize 后，新事件将被拒绝。在增加大小之前，请先检查为什么没有快速消费 — 通常问题不在引擎，而是规则本身太重。

处理失败的方法

异常	何时
EPCompileException	EQL 编译失败 (validate · deploy)
CEPClientClosedException	在 close() 之后使用
IllegalArgumentException	参数错误 — 空的规则名称、空的 EQL
RuntimeException	服务器 4xx/5xx (包括认证失败 401)、网络错误

相关文档

Data Gateway 客户端 · CEP 组件 · EQL 用法