본문으로 건너뛰기

CEP 클라이언트

CEPClient 는 실시간 규칙 엔진에 붙는 자바 라이브러리입니다. 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 는 내부적으로 모아 보냅니다. 초당 수만 건을 밀어 넣을 때 조정합니다.

파라미터기본값뜻
shardCount4flush 워커 스레드 수
queueSize600,000전체 큐 크기
flushTriggerSize8,000이 크기에 닿으면 flush
maxInflightBatches64동시에 진행할 수 있는 flush 수
batchIntervalMs50시간 기반 flush 간격
skipResulttrue결과를 기다리지 않음(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 를 넘기면 새 이벤트가 거절됩니다. 늘리기 전에 왜 안 빠지는지를 먼저 보십시오 — 대개 엔진이 아니라 규칙이 무겁습니다.

실패를 다루는 법​

예외언제
EPCompileExceptionEQL 컴파일 실패 (validate · deploy)
CEPClientClosedExceptionclose() 뒤에 사용
IllegalArgumentException잘못된 인자 — 빈 규칙 이름, 빈 EQL
RuntimeException서버 4xx/5xx(인증 실패 401 포함), 네트워크 오류

함께 보기​

Data Gateway 클라이언트 · CEP 컴포넌트 · EQL 사용법