7.4 키 표현식(Key Expressions)의 이해와 활용

7.4 키 표현식(Key Expressions)의 이해와 활용

Zenoh 세계에서 데이터를 찾는 방법은 IP 주소도, 메모리 포인터도 아니다. 오직 Key Expression (경로 표현식) 하나로 모든 트래픽이 분류되고 라우팅된다. REST API의 URL 경로(Uniform Resource Locator)와 비슷해 보이지만, 단수(Single)가 아니라 복수(Multiple) 의 매칭을 동시에 처리하는 정규표현식 수준의 권력을 가졌다.

이 장에서는 Python 환경에서 robot/1/battery 같은 평범한 문자열이 어떻게 Zenoh 코어 라우팅 엔진의 이진 트리(Binary Tree)로 맵핑(Mapping)되는지, 그리고 * 나 ** 와일드카드를 써서 전 세계에 흩어진 로봇 천 대의 센서 데이터를 파이썬 딕셔너리로 단 한 줄에 긁어오는 무결점 데이터 색인 기법을 파헤친다.

1. 키 표현식의 기본 구조와 Python 문자열 매핑

Python에서는 C++이나 Rust처럼 귀찮은 바이트 캐스팅을 피하고, 100% 네이티브 Python String 타입 하나로 모든 키를 쏘고 받는다.

1.0.1 [Runbook] URI 기반 토픽 설계 전술

Zenoh의 Key Expression은 REST 아키텍처의 자원(Resource) 설계 사상을 충실히 따른다. 알파벳, 숫자, 하이픈(-), 언더바(_), 그리고 계층을 분절하는 슬래시(/) 의 조합이다.

1. [규칙] 절대 경로 체계

  • 모든 경로는 논리적 그룹을 좌측에 둔다 (예: domain / region / class / instance / device).
  • korea/seoul/factory_a/robot_01/battery

2. [퍼포먼스 튜닝] 파이썬 문자열 vs Zenoh KeyExpr 객체
파이썬 스크립트에서 이렇게 데이터를 쏠 수 있다.

## 방법 A: 평범한 파이썬 String 편하게 던지기
session.put("room/1/temp", "25C")

이 코드는 사실, 매번 파이썬 String이 Rust단으로 넘어가 KeyExpr 객체로 파싱되는 오버헤드(문법 검사 등)를 낳는다. 만약 초당 1,000번 호출되는 딥러닝 카메라 추론 루프라면?
키 객체를 미리 컴파일(캐싱) 해두는 것이 고수의 필수 전술이다.

import zenoh

session = zenoh.open()

## [최적화] 프로그램 기동 시, 미리 파이썬 문자열을 컴파일된 Zenoh 식별자 객체로 구워둔다.
## 이 행위는 정규표현식의 re.compile() 과 완전히 똑같은 이치다.
compiled_key = zenoh.KeyExpr("room/1/temp")

def fast_loop():
    for _ in range(1000):
        # 파싱 시간을 아끼고 곧장 바이너리로 밀어버린다.
        session.put(compiled_key, "25C")

이 zenoh.KeyExpr 클래스 활용 하나만으로 파이썬 통신 성능이 최대 20% 상승한다.

2. 와일드카드(*, ``)를 활용한 계층적 패턴 매칭 기법

백엔드 파이썬 서버가 수천 대의 단말을 관리할 때, 모든 단말의 IP나 Key를 List에 담아놓고 For-loop로 긁어모으는 짓(Polling)은 최악의 안티 패턴이다.
Zenoh의 강력함은, 클라이언트가 던지는 와일드카드 질의(Wildcard Query) 가 라우터 코어 단에서 순식간에 복제(Fork)되어 전 세계 하위 노드에게 맹렬하게 뻗어 나가는 데 있다.

2.0.1 [Runbook] 시공간 제압 타겟팅 전술

  • * : 하나의 계층(Depth) 내에서 글자 수 무관 매칭
  • ** : 0개 이상의 모든 하위 계층(Depth) 무차별 타격 매칭

전술 1. 층위 제한 타격 (별 1개)

## 'robot/1/battery', 'robot/999/battery' 등 모두 캐치함.
## 단, 'robot/1/arm/battery' 처럼 깊이가 다르면 잡지 못함.
sub = session.declare_subscriber("robot/*/battery", lambda s: print(s.payload))

전술 2. 전방위 카펫 폭격 (별 2개)
공장 전체의 모든 센서와 로그를 닥치는 대로 긁어모으는 무지막지한 권력이다.

## 'factoryA' 산하에 있는 모든 depth의 모든 데이터를 가로챈다.
## factoryA/logs, factoryA/robot/arm/temp, 심지어 factoryA 본체까지!
sub = session.declare_subscriber("factoryA/**", lambda s: print(s.key_expr))

전술 3. [고급] 도메인 프리픽스(Prefix) 제한 폭격
특정 접두어로 시작되는 센서들만 솎아내고 싶을 때 파이썬의 startswith 코드를 짤 필요가 없다.

## 'temp_in', 'temp_out', 'temperature' 등 temp 로 시작하는 모든 것을 일괄 수집
sub = session.declare_subscriber("sensor/temp*", lambda s: print(s.payload))

이 와일드카드 토폴로지 구조는 RabbitMQ의 Topic Exchange 나 Kafka의 정규식 리스너를 한참 상회할 만큼 가볍고, 라우터가 아예 트래픽 방향을 꺾어 불필요한 네트워크 지점을 지나치게(Bypass) 만들 정도로 성능 튜닝의 핵심 지표가 된다.

3. 키 표현식 검증, 교차 확인(Intersection) 변환 유틸리티

거대한 시스템에서는 내가 건드리는 경로(robot/*/temp)가 혹시 다른 부서가 쓰고 있는 경로(robot/admin/**)와 겹쳐서 의도치 않은 시스템 셧다운 패킷을 수신(Intersection)하게 되는 건 아닌지 사전에 계산해야 한다.
파이썬 Zenoh 라이브러리는 이를 위해 기하학적 집합론(Set Theory)에 입각한 경로 비교 알고리즘을 무료로 제공한다.

3.0.1 [Runbook] 데이터 영토(Domain) 무결성 검사 전술

1. 교집합(Intersection) 방파제 검증

“A경로로 쏘면 B구독자가 낚아채게 될까?“를 라우터 연결도 없이 오프라인 파이썬 객체만으로 증명(Assert)한다.

import zenoh

def validate_topology():
    expr_sub = zenoh.KeyExpr("team_a/drone/*/camera")
    expr_pub = zenoh.KeyExpr("team_a/drone/5/camera")
    expr_other = zenoh.KeyExpr("team_b/drone/5/camera")
    
    # 1. 포섭 여부 (A가 B를 포함하는가?)
    if expr_sub.includes(expr_pub):
        print("[통과] pub가 쏘면 sub가 정확히 받아낼 수 있습니다.")
        
    # 2. 교집합 발생 여부 (우연히 겹치는 영역이 있는가?)
    if expr_sub.intersects(expr_other):
        print("[경고] 남의 팀 데이터가 혼입될 위험이 있습니다!")
    else:
        print("[안전] 영토가 완벽히 분리되어 있습니다.")
        
## 런타임에 에러가 터지기 전(Init Phase)에 이 검증 방어막을 친다.
validate_topology()

이 유틸리티는 데브옵스 CI/CD 파이프라인에서 “개발자가 Commit한 설정 파일의 Topic 명이, 프로덕션의 중요 보안 Topic 영역(admin/**)을 침범하지 않는지“를 사전에 쳐내는 자동화 테스트(Unit Test)를 작성할 때 진정한 가치를 발휘한다.

4. 동적 키 표현식 생성 및 포매팅

10,000대의 공장 IoT 센서는 코드 시작부터 죽을 때까지 단 1개의 고정된 이름표(KeyExpr)를 달고 돌지 않는다. “방에 불이 꺼지면 이름이 room/1/temp 에 room/1/dark/temp로 바뀐다“는 식의 동적 생성(Dynamic Formatter) 시스템이 현장의 99%를 차지한다.

이 포매팅 과정에서 파이썬의 위력이 빛을 발한다. C++이나 Rust라면 sprintf 지옥에 빠졌을 텐데, 파이썬은 f-String 하나로 모든 것을 정리한다.

4.0.1 [Runbook] 다차원 식별코드 동적 캐스팅 전술

import zenoh

session = zenoh.open()

class DynamicSensor:
    def __init__(self, room_id, sensor_type):
        self.room = room_id
        self.type = sensor_type
        # 1. f-string 을 통해 인간 친화적으로 변수를 주입한다.
        self.topic_base = f"building_a/room_{self.room}/{self.type}"
        self.pub = session.declare_publisher(self.topic_base)

    def report(self, is_warning, val):
        # 2. 전송 순간에 뒷부분 꼬리표(Tail)만 동적으로 치환한다.
        # 주의: pub.put() 에 경로를 덮어쓰려면 선언 시에 넓은 영토가 필요하지만,
        # 세션 레벨에서 session.put(dynamic_key, val) 로 때리는 것이 안전하다.
        
        state_str = "alert" if is_warning else "normal"
        
        # 완전 동적 포매팅 폭격
        dynamic_target = f"{self.topic_base}/{state_str}"
        
        session.put(dynamic_target, str(val))
        print(f"동적 타격 완료: {dynamic_target}")

## 1번방 센서 객체 
sensor_1 = DynamicSensor(1, "temp")
sensor_1.report(False, 22.5) # building_a/room_1/temp/normal 로 전송됨
sensor_1.report(True, 45.0)  # building_a/room_1/temp/alert 로 전송됨

만약 이런 동적 문자열 결합 과정이 너무 잦아 오버헤드가 발생한다면, Zenoh 파이썬 바인딩은 파이썬 객체 생성 부하를 줄이기 위한 딕셔너리(dict) 매핑 기법도 지원한다. 하지만 가독성과 런타임 효율성 트레이드오프에서 f-string은 100만 TPS (Transactions Per Second) 미만의 구간에서는 언제나 최고의 포매팅 도구(Silver Bullet)로 군림한다.