订阅系统：实时数据变更通知
Tags: #Subscription #Realtime #ChangeNotification #WebSocket #EventDriven #智策平台
Coomia发布于 2025年8月1日7 分钟阅读
分享本文Twitter / X
“系列：S3 数据基座 · 第 22 篇 | 难度：高级 | 阅读时间：20 分钟
订阅系统：实时数据变更通知
Tags: #Subscription #Realtime #ChangeNotification #WebSocket #EventDriven #智策平台
#TL;DR
coomia-dip 的订阅系统让用户和应用程序能够订阅 Ontology 数据的变更事件——当特定 Entity Type 的属性发生变化、关系被创建或删除、指标超过阈值时，系统自动推送通知。本文完整解析订阅系统的架构设计，包括变更事件捕获（基于 Iceberg Changelog 和 Flink CDC）、订阅规则定义与匹配引擎、多通道推送（WebSocket、Server-Sent Events、Webhook、消息队列）、事件去重与排序保障、订阅性能优化和典型应用场景。
#1. 订阅系统架构
#1.1 整体架构
Code
订阅系统架构：
┌──────────────────────────────────────────────────┐
│ Data Sources │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Flink CDC│ │ Iceberg │ │ World │ │
│ │ Events │ │ Changelog│ │ Transform│ │
│ └──────────┘ └──────────┘ └──────────┘ │
└──────────────────────┬───────────────────────────┘
│ Events
▼
┌──────────────────────────────────────────────────┐
│ Event Processing Pipeline │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Dedup │→│ Enrich │→│ Match │ │
│ │ & Order │ │ (Context)│ │ (Rules) │ │
│ └──────────┘ └──────────┘ └──────────┘ │
└──────────────────────┬───────────────────────────┘
│ Matched Events
▼
┌──────────────────────────────────────────────────┐
│ Notification Dispatch │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌────┐ │
│ │WebSocket │ │ SSE │ │ Webhook │ │ MQ │ │
│ └──────────┘ └──────────┘ └──────────┘ └────┘ │
└──────────────────────────────────────────────────┘
#1.2 事件类型
Code
Ontology 变更事件类型：
┌─────────────────┬──────────────────────────────────┐
│ 事件类型 │ 触发条件 │
├─────────────────┼──────────────────────────────────┤
│ ENTITY_CREATED │ 新实体创建 │
│ ENTITY_UPDATED │ 实体属性变更 │
│ ENTITY_DELETED │ 实体删除 │
│ EDGE_CREATED │ 新关系创建 │
│ EDGE_DELETED │ 关系删除 │
│ METRIC_CHANGED │ 指标值变更 │
│ METRIC_THRESHOLD │ 指标超过/低于阈值 │
│ TYPE_SCHEMA_CHG │ Entity Type Schema 变更 │
│ BRANCH_MERGED │ Nessie 分支合并 │
│ TRANSFORM_DONE │ World Transform 执行完成 │
└─────────────────┴──────────────────────────────────┘
#2. 订阅规则定义
#2.1 订阅 API
Python
from onto_subscription import Subscription, EventFilter, Channel
# 订阅示例 1：监控高风险客户变更
sub1 = Subscription(
name="high-risk-customer-watch",
description="监控高风险客户的任何属性变更",
filters=[
EventFilter(
event_type="ENTITY_UPDATED",
entity_type="Customer",
property_filter={"risk_level": "HIGH"},
)
],
channels=[
Channel.webhook("https://alert.company.com/risk"),
Channel.email("risk-team@company.com"),
],
debounce_seconds=30,
)
# 订阅示例 2：设备离线告警
sub2 = Subscription(
name="device-offline-alert",
filters=[
EventFilter(
event_type="ENTITY_UPDATED",
entity_type="Device",
property_changes={"status": {"from": "online", "to": "offline"}},
)
],
channels=[
Channel.websocket(room="ops-dashboard"),
Channel.dingtalk(robot_url="https://oapi.dingtalk.com/robot/send?..."),
],
debounce_seconds=0, # 立即通知，不防抖
)
# 订阅示例 3：指标阈值告警
sub3 = Subscription(
name="cpu-threshold-alert",
filters=[
EventFilter(
event_type="METRIC_THRESHOLD",
entity_type="Server",
metric_name="cpu_utilization",
threshold={"operator": ">", "value": 90},
)
],
channels=[
Channel.pagerduty(service_key="xxx"),
],
cooldown_minutes=15, # 15分钟内不重复告警
)
#2.2 规则匹配引擎
Python
class SubscriptionMatcher:
"""订阅规则匹配引擎"""
def __init__(self):
self._subscriptions: list[Subscription] = []
self._index: dict[str, list[Subscription]] = {}
def register(self, subscription: Subscription):
self._subscriptions.append(subscription)
for f in subscription.filters:
key = f"{f.event_type}:{f.entity_type}"
self._index.setdefault(key, []).append(subscription)
def match(self, event: OntologyEvent) -> list[MatchResult]:
key = f"{event.type}:{event.entity_type}"
candidates = self._index.get(key, [])
matches = []
for sub in candidates:
for filter_rule in sub.filters:
if self._matches_filter(event, filter_rule):
matches.append(MatchResult(
subscription=sub,
event=event,
matched_filter=filter_rule,
))
break
return matches
def _matches_filter(self, event: OntologyEvent, f: EventFilter) -> bool:
if f.property_filter:
for prop, expected in f.property_filter.items():
actual = event.entity_properties.get(prop)
if actual != expected:
return False
if f.property_changes:
for prop, change in f.property_changes.items():
if prop not in event.changed_properties:
return False
if 'from' in change and event.old_values.get(prop) != change['from']:
return False
if 'to' in change and event.new_values.get(prop) != change['to']:
return False
if f.threshold:
metric_value = event.metric_value
op = f.threshold['operator']
threshold = f.threshold['value']
if op == '>' and not (metric_value > threshold):
return False
if op == '<' and not (metric_value < threshold):
return False
return True
#3. 通知推送
#3.1 WebSocket 推送
Python
class WebSocketNotifier:
"""WebSocket 实时推送"""
def __init__(self):
self._connections: dict[str, set[WebSocket]] = {}
async def on_connect(self, ws: WebSocket, room: str):
self._connections.setdefault(room, set()).add(ws)
async def on_disconnect(self, ws: WebSocket, room: str):
self._connections.get(room, set()).discard(ws)
async def notify(self, room: str, event: NotificationEvent):
connections = self._connections.get(room, set())
dead = set()
for ws in connections:
try:
await ws.send_json(event.to_dict())
except ConnectionClosed:
dead.add(ws)
for ws in dead:
connections.discard(ws)
#3.2 Webhook 推送
Python
class WebhookNotifier:
"""Webhook 推送（带重试）"""
async def notify(self, url: str, event: NotificationEvent):
payload = event.to_dict()
for attempt in range(3):
try:
async with aiohttp.ClientSession() as session:
resp = await session.post(
url,
json=payload,
headers={
'Content-Type': 'application/json',
'X-Onto-Event-Type': event.type,
'X-Onto-Signature': self._sign(payload),
},
timeout=aiohttp.ClientTimeout(total=10),
)
if resp.status < 300:
return
if resp.status >= 500:
await asyncio.sleep(2 ** attempt)
continue
return # 4xx 不重试
except Exception:
await asyncio.sleep(2 ** attempt)
await self._dead_letter_queue.put(event)
#4. 事件去重与排序
Python
class EventDeduplicator:
"""事件去重和排序"""
def __init__(self, window_seconds: int = 60):
self._seen: dict[str, datetime] = {}
self._window = timedelta(seconds=window_seconds)
def deduplicate(self, events: list[OntologyEvent]) -> list[OntologyEvent]:
unique = []
for event in sorted(events, key=lambda e: e.timestamp):
key = f"{event.entity_id}:{event.type}:{event.property_hash}"
last_seen = self._seen.get(key)
if last_seen and event.timestamp - last_seen < self._window:
continue
self._seen[key] = event.timestamp
unique.append(event)
self._cleanup_expired()
return unique
#5. 性能优化
Code
订阅系统性能指标：
┌─────────────────┬──────────┬──────────┐
│ 指标 │ 目标 │ 实际 │
├─────────────────┼──────────┼──────────┤
│ 事件到通知延迟 │ < 1s │ 200ms │
│ 并发订阅数 │ 10,000 │ 15,000 │
│ 事件吞吐量 │ 10K/s │ 25K/s │
│ WebSocket 连接 │ 5,000 │ 8,000 │
│ 匹配延迟 │ < 1ms │ 0.3ms │
└─────────────────┴──────────┴──────────┘
#6. 测试策略
Python
class TestSubscriptionSystem:
async def test_property_change_triggers_notification(self):
sub = Subscription(
filters=[EventFilter(
event_type="ENTITY_UPDATED",
entity_type="Device",
property_changes={"status": {"to": "offline"}}
)],
channels=[Channel.memory_queue("test")]
)
matcher.register(sub)
event = OntologyEvent(
type="ENTITY_UPDATED",
entity_type="Device",
entity_id="d1",
changed_properties=["status"],
old_values={"status": "online"},
new_values={"status": "offline"},
)
matches = matcher.match(event)
assert len(matches) == 1
async def test_debounce_prevents_flooding(self):
sub = Subscription(
filters=[EventFilter(event_type="ENTITY_UPDATED", entity_type="Sensor")],
channels=[Channel.memory_queue("test")],
debounce_seconds=5,
)
# 发送 100 个事件在 1 秒内
for i in range(100):
await process_event(make_event(f"s{i}"))
# 应只收到 1 个通知（被防抖）
notifications = get_queue("test")
assert len(notifications) <= 10
async def test_deduplication(self):
dedup = EventDeduplicator(window_seconds=10)
events = [make_event("e1", ts=i) for i in range(5)] # 相同事件 5 次
unique = dedup.deduplicate(events)
assert len(unique) == 1
#Key Takeaways
-
订阅系统将 Ontology 从"被动查询"升级为"主动推送"：用户和应用不需要轮询数据变更，系统在数据变化时自动通知。
-
灵活的事件过滤支持细粒度订阅：支持按 Entity Type、属性值、属性变更方向、指标阈值等维度精确订阅。
-
多通道推送覆盖不同场景：WebSocket 用于实时仪表盘，Webhook 用于系统集成，邮件/钉钉用于运维告警。
-
事件去重和防抖防止通知洪泛：去重窗口和防抖机制确保高频变更场景下不会产生通知风暴。
-
事件到通知的端到端延迟 < 1 秒：从数据变更到通知推送在亚秒级完成，满足实时场景需求。
#Next Article
下一篇 S3-23《数据导出：多格式批量与流式输出》 将展示如何从 Ontology 中导出数据到 CSV、Parquet、Excel 等多种格式。
Tags: #Subscription #Realtime #ChangeNotification #WebSocket #Webhook #EventDriven #Debounce #智策平台 #coomia-dip #数据基座