通知引擎：9 种通知渠道的统一抽象
企业决策的最后一环是将结果通知到正确的人。coomia-dip 的 NotificationEngine 通过渠道适配器模式统一管理 9 种通知渠道——Email、SMS、Webhook、Slack、钉钉、企业微信、飞书、站内信和 Push。本文深入解析通知引擎的适配器架构、模板渲染引擎、批量投递优化、用户偏好路由、投递状态追踪以及防骚扰策略，帮助你构建可靠且用户友好的企业通知系统。
Coomia发布于 2025年9月8日13 分钟阅读
分享本文Twitter / X
“系列：S5 智能决策 · 第 16 篇 | 难度：高级 | 阅读时间：20 分钟
通知引擎：9 种通知渠道的统一抽象
#TL;DR
企业决策的最后一环是将结果通知到正确的人。coomia-dip 的 NotificationEngine 通过渠道适配器模式统一管理 9 种通知渠道——Email、SMS、Webhook、Slack、钉钉、企业微信、飞书、站内信和 Push。本文深入解析通知引擎的适配器架构、模板渲染引擎、批量投递优化、用户偏好路由、投递状态追踪以及防骚扰策略，帮助你构建可靠且用户友好的企业通知系统。
#1. 企业通知的挑战
#1.1 渠道碎片化
Code
传统通知架构：
┌────────┐ ┌──────────┐
│ 审批模块 │───→│ 邮件服务 │ API-1
└────────┘ └──────────┘
┌────────┐ ┌──────────┐
│ 告警模块 │───→│ 短信网关 │ API-2
└────────┘ └──────────┘
┌────────┐ ┌──────────┐
│ 协作模块 │───→│ 钉钉机器人│ API-3
└────────┘ └──────────┘
问题：
- 每个模块独立对接渠道
- 模板格式不统一
- 用户偏好无法集中管理
- 防骚扰策略各自为政
#1.2 统一通知的目标
| 目标 | 说明 |
|---|---|
| 渠道无关 | 业务代码不关心具体用哪个渠道 |
| 模板统一 | 一套模板引擎，跨渠道渲染 |
| 偏好路由 | 根据用户偏好自动选择渠道 |
| 防骚扰 | 全局频率控制，避免通知轰炸 |
| 可追溯 | 每条通知有完整的投递日志 |
| 降级容错 | 主渠道不可用时自动切换备用渠道 |
#2. 通知引擎架构
#2.1 整体架构
Code
┌─────────────────────────────────────────────────────────────┐
│ NotificationEngine │
│ │
│ ┌──────────┐ ┌──────────────┐ ┌──────────────────┐ │
│ │ Notify │───→│ Template │───→│ Preference │ │
│ │ Request │ │ Renderer │ │ Router │ │
│ └──────────┘ └──────────────┘ └────────┬─────────┘ │
│ │ │
│ ┌────────────────────────────┤ │
│ ▼ ▼ ▼ │
│ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │
│ │ Rate │ │ Dedup │ │ Channel │ │
│ │ Limiter │ │ Filter │ │ Dispatcher │ │
│ └──────────┘ └──────────┘ └──────┬───────┘ │
│ │ │
│ ┌──────────────────────────────────────┤ │
│ ▼ ▼ ▼ ▼ ▼ │
│ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────────┐ │
│ │Email │ │ SMS │ │Slack │ │Ding- │ │ 更多... │ │
│ │Adapt.│ │Adapt.│ │Adapt.│ │Talk │ │ │ │
│ └──────┘ └──────┘ └──────┘ └──────┘ └──────────┘ │
│ │
│ ┌──────────────────────────────────────────────────┐ │
│ │ Delivery Tracker │ Audit Log │ Metrics │ │
│ └──────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
#2.2 核心数据模型
Python
class NotificationChannel(str, Enum):
EMAIL = "email"
SMS = "sms"
WEBHOOK = "webhook"
SLACK = "slack"
DINGTALK = "dingtalk"
WECHAT_WORK = "wechat_work"
FEISHU = "feishu"
IN_APP = "in_app"
PUSH = "push"
class NotificationPriority(str, Enum):
CRITICAL = "critical" # 立即投递，绕过限频
HIGH = "high" # 尽快投递
NORMAL = "normal" # 正常投递
LOW = "low" # 批量合并投递
class NotificationRequest(BaseModel):
notification_id: str = Field(
default_factory=lambda: str(uuid.uuid4())
)
channel: NotificationChannel | None = None # None = 使用偏好
template_id: str
recipients: list[Recipient]
variables: dict[str, Any] = Field(default_factory=dict)
priority: NotificationPriority = NotificationPriority.NORMAL
metadata: dict[str, Any] = Field(default_factory=dict)
schedule_at: datetime | None = None # 定时发送
idempotency_key: str | None = None
triggered_by: str = "system"
class Recipient(BaseModel):
user_id: str | None = None
email: str | None = None
phone: str | None = None
channel_id: str | None = None # Slack channel, DingTalk group
metadata: dict[str, Any] = Field(default_factory=dict)
class DeliveryRecord(BaseModel):
record_id: str = Field(
default_factory=lambda: str(uuid.uuid4())
)
notification_id: str
recipient: Recipient
channel: NotificationChannel
status: str # queued | sending | delivered | failed | bounced
content: str
sent_at: datetime | None = None
delivered_at: datetime | None = None
error: str | None = None
external_id: str | None = None # 外部系统消息 ID
#3. 渠道适配器
#3.1 适配器接口
Python
class ChannelAdapter(ABC):
"""渠道适配器基类"""
@property
@abstractmethod
def channel(self) -> NotificationChannel:
...
@abstractmethod
async def send(
self, recipient: Recipient,
content: RenderedContent,
metadata: dict,
) -> DeliveryResult:
...
@abstractmethod
async def send_batch(
self, messages: list[tuple[Recipient, RenderedContent]],
metadata: dict,
) -> list[DeliveryResult]:
...
@abstractmethod
async def health_check(self) -> bool:
...
def supports_rich_content(self) -> bool:
return False
def max_batch_size(self) -> int:
return 1
#3.2 各渠道实现
Email 适配器
Python
class EmailAdapter(ChannelAdapter):
"""Email 渠道适配器"""
@property
def channel(self) -> NotificationChannel:
return NotificationChannel.EMAIL
def __init__(self, smtp_config: SMTPConfig):
self.smtp = smtp_config
async def send(
self, recipient: Recipient,
content: RenderedContent,
metadata: dict,
) -> DeliveryResult:
msg = EmailMessage()
msg["Subject"] = content.subject or content.title
msg["From"] = self.smtp.from_address
msg["To"] = recipient.email
msg["Message-ID"] = make_msgid()
if content.html:
msg.set_content(content.text)
msg.add_alternative(content.html, subtype="html")
else:
msg.set_content(content.text)
# 附件
for attachment in content.attachments:
msg.add_attachment(
attachment.data,
maintype=attachment.mime_type.split("/")[0],
subtype=attachment.mime_type.split("/")[1],
filename=attachment.filename,
)
async with aiosmtplib.SMTP(
hostname=self.smtp.host,
port=self.smtp.port,
use_tls=self.smtp.use_tls,
) as smtp:
await smtp.login(self.smtp.username, self.smtp.password)
await smtp.send_message(msg)
return DeliveryResult(
success=True,
channel=self.channel,
external_id=msg["Message-ID"],
)
def supports_rich_content(self) -> bool:
return True
def max_batch_size(self) -> int:
return 100
钉钉适配器
Python
class DingTalkAdapter(ChannelAdapter):
"""钉钉渠道适配器"""
@property
def channel(self) -> NotificationChannel:
return NotificationChannel.DINGTALK
async def send(
self, recipient: Recipient,
content: RenderedContent,
metadata: dict,
) -> DeliveryResult:
# 工作通知消息
if recipient.user_id:
return await self._send_work_notice(
recipient, content
)
# 群消息（通过 Webhook 机器人）
if recipient.channel_id:
return await self._send_group_message(
recipient, content
)
raise ValueError("DingTalk needs user_id or channel_id")
async def _send_work_notice(
self, recipient: Recipient,
content: RenderedContent,
) -> DeliveryResult:
payload = {
"agent_id": self.config.agent_id,
"userid_list": recipient.user_id,
"msg": {
"msgtype": "markdown",
"markdown": {
"title": content.title,
"text": content.text,
},
},
}
token = await self._get_access_token()
async with httpx.AsyncClient() as client:
resp = await client.post(
"https://oapi.dingtalk.com/topapi/message"
"/corpconversation/asyncsend_v2",
params={"access_token": token},
json=payload,
)
data = resp.json()
return DeliveryResult(
success=data.get("errcode") == 0,
channel=self.channel,
external_id=str(data.get("task_id")),
error=data.get("errmsg") if data.get("errcode") != 0 else None,
)
飞书适配器
Python
class FeishuAdapter(ChannelAdapter):
"""飞书渠道适配器"""
@property
def channel(self) -> NotificationChannel:
return NotificationChannel.FEISHU
async def send(
self, recipient: Recipient,
content: RenderedContent,
metadata: dict,
) -> DeliveryResult:
# 构建消息卡片
card = {
"config": {"wide_screen_mode": True},
"header": {
"title": {
"tag": "plain_text",
"content": content.title,
},
"template": self._get_template_color(
metadata.get("priority", "normal")
),
},
"elements": [
{
"tag": "markdown",
"content": content.text,
},
],
}
# 添加操作按钮
if content.actions:
card["elements"].append({
"tag": "action",
"actions": [
{
"tag": "button",
"text": {
"tag": "plain_text",
"content": action.label,
},
"url": action.url,
"type": action.style or "default",
}
for action in content.actions
],
})
payload = {
"receive_id": recipient.user_id,
"msg_type": "interactive",
"content": json.dumps(card),
}
token = await self._get_tenant_access_token()
async with httpx.AsyncClient() as client:
resp = await client.post(
"https://open.feishu.cn/open-apis/im/v1/messages",
params={"receive_id_type": "user_id"},
headers={"Authorization": f"Bearer {token}"},
json=payload,
)
data = resp.json()
return DeliveryResult(
success=data.get("code") == 0,
channel=self.channel,
external_id=data.get("data", {}).get("message_id"),
error=data.get("msg") if data.get("code") != 0 else None,
)
#3.3 渠道能力矩阵
| 渠道 | 富文本 | 按钮 | 附件 | 批量 | 已读回执 | 双向互动 |
|---|---|---|---|---|---|---|
| HTML | No | Yes | 100 | Opt. | No | |
| SMS | No | No | No | 1000 | No | No |
| Webhook | JSON | N/A | No | 50 | N/A | N/A |
| Slack | Markdown | Yes | Yes | 50 | Yes | Yes |
| DingTalk | Markdown | Yes | Yes | 100 | Yes | Yes |
| WeCom | Markdown | Yes | Yes | 100 | No | Yes |
| Feishu | Markdown | Yes | Yes | 100 | Yes | Yes |
| In-App | HTML | Yes | Yes | N/A | Auto | Yes |
| Push | Short | Yes | No | 1000 | No | No |
#4. 模板引擎
#4.1 多格式模板
Python
class NotificationTemplate(BaseModel):
template_id: str
name: str
description: str
# 多格式内容
subject: str | None = None # Email 主题
title: str # 标题
text_template: str # 纯文本
html_template: str | None = None # HTML（Email）
markdown_template: str | None = None # Markdown（IM）
short_template: str | None = None # 短文本（SMS/Push）
# 渠道特定模板
channel_overrides: dict[str, dict] = Field(
default_factory=dict
)
# 国际化
i18n: dict[str, dict] = Field(default_factory=dict)
class TemplateRenderer:
"""模板渲染引擎"""
async def render(
self, template_id: str,
variables: dict,
channel: NotificationChannel,
locale: str = "zh-CN",
) -> RenderedContent:
template = await self.template_store.get(template_id)
# 选择语言版本
if locale in template.i18n:
template_data = template.i18n[locale]
else:
template_data = template.model_dump()
# 渠道覆盖
if channel.value in template.channel_overrides:
template_data.update(
template.channel_overrides[channel.value]
)
# 渲染模板
env = jinja2.Environment(
undefined=jinja2.StrictUndefined,
autoescape=True,
)
text = env.from_string(
template_data.get("text_template", "")
).render(**variables)
# 根据渠道选择格式
match channel:
case NotificationChannel.EMAIL:
html = env.from_string(
template_data.get("html_template", "")
).render(**variables)
return RenderedContent(
title=template_data.get("title", ""),
subject=template_data.get("subject", ""),
text=text,
html=html,
)
case (NotificationChannel.SLACK
| NotificationChannel.DINGTALK
| NotificationChannel.FEISHU
| NotificationChannel.WECHAT_WORK):
md = env.from_string(
template_data.get("markdown_template", text)
).render(**variables)
return RenderedContent(
title=template_data.get("title", ""),
text=md,
)
case NotificationChannel.SMS | NotificationChannel.PUSH:
short = env.from_string(
template_data.get("short_template", text[:200])
).render(**variables)
return RenderedContent(
title=template_data.get("title", ""),
text=short,
)
case _:
return RenderedContent(
title=template_data.get("title", ""),
text=text,
)
#5. 用户偏好路由
#5.1 偏好模型
Python
class UserNotificationPreference(BaseModel):
user_id: str
default_channel: NotificationChannel = (
NotificationChannel.IN_APP
)
channel_preferences: dict[str, NotificationChannel] = Field(
default_factory=dict
)
quiet_hours: QuietHours | None = None
disabled_categories: list[str] = Field(default_factory=list)
digest_enabled: bool = False
digest_interval: str = "daily" # "hourly" | "daily" | "weekly"
class QuietHours(BaseModel):
start: str = "22:00" # HH:MM
end: str = "08:00"
timezone: str = "Asia/Shanghai"
override_critical: bool = True # critical 消息忽略安静时段
class PreferenceRouter:
"""根据用户偏好路由通知渠道"""
async def resolve_channel(
self, recipient: Recipient,
request: NotificationRequest,
) -> NotificationChannel:
# 1. 如果请求指定了渠道，直接使用
if request.channel:
return request.channel
# 2. 查询用户偏好
prefs = await self.pref_store.get(recipient.user_id)
if not prefs:
return NotificationChannel.IN_APP
# 3. 按通知类别查找偏好
category = request.metadata.get("category", "default")
if category in prefs.channel_preferences:
channel = prefs.channel_preferences[category]
else:
channel = prefs.default_channel
# 4. 安静时段检查
if prefs.quiet_hours and self._is_quiet_time(
prefs.quiet_hours
):
if (request.priority != NotificationPriority.CRITICAL
or not prefs.quiet_hours.override_critical):
return NotificationChannel.IN_APP
return channel
#6. 防骚扰策略
#6.1 频率限制
Python
class NotificationRateLimiter:
"""通知频率限制器"""
LIMITS = {
NotificationChannel.EMAIL: {
"per_hour": 10,
"per_day": 50,
},
NotificationChannel.SMS: {
"per_hour": 3,
"per_day": 10,
},
NotificationChannel.PUSH: {
"per_hour": 5,
"per_day": 20,
},
}
async def check(
self, recipient: Recipient,
channel: NotificationChannel,
priority: NotificationPriority,
) -> RateLimitResult:
# Critical 消息绕过限频
if priority == NotificationPriority.CRITICAL:
return RateLimitResult(allowed=True)
limits = self.LIMITS.get(channel)
if not limits:
return RateLimitResult(allowed=True)
key = f"notify:{recipient.user_id}:{channel.value}"
# 检查小时限制
hour_count = await self.counter.get(f"{key}:hour")
if hour_count >= limits["per_hour"]:
return RateLimitResult(
allowed=False,
reason=f"Hourly limit ({limits['per_hour']}) reached",
retry_after_seconds=3600,
)
# 检查日限制
day_count = await self.counter.get(f"{key}:day")
if day_count >= limits["per_day"]:
return RateLimitResult(
allowed=False,
reason=f"Daily limit ({limits['per_day']}) reached",
retry_after_seconds=86400,
)
# 计数
await self.counter.increment(f"{key}:hour", ttl=3600)
await self.counter.increment(f"{key}:day", ttl=86400)
return RateLimitResult(allowed=True)
#6.2 通知合并（Digest）
Python
class NotificationDigest:
"""低优先级通知合并为摘要"""
async def maybe_digest(
self, request: NotificationRequest
) -> NotificationRequest | None:
"""
如果用户开启了摘要模式且通知优先级为 LOW，
将通知加入摘要队列而不是立即发送。
"""
if request.priority != NotificationPriority.LOW:
return request
for recipient in request.recipients:
prefs = await self.pref_store.get(recipient.user_id)
if prefs and prefs.digest_enabled:
await self.digest_queue.add(
user_id=recipient.user_id,
notification=request,
interval=prefs.digest_interval,
)
return None # 不立即发送
return request # 未开启摘要，正常发送
async def flush_digest(self, user_id: str) -> None:
"""发送摘要邮件"""
notifications = await self.digest_queue.drain(user_id)
if not notifications:
return
digest_request = NotificationRequest(
template_id="system_digest",
channel=NotificationChannel.EMAIL,
recipients=[Recipient(user_id=user_id)],
variables={
"count": len(notifications),
"items": [
{
"title": n.variables.get("title", ""),
"summary": n.variables.get("summary", ""),
"time": n.metadata.get("created_at", ""),
}
for n in notifications
],
},
priority=NotificationPriority.LOW,
)
await self.engine.send(digest_request)
#7. 渠道降级与容错
#7.1 降级链
Code
渠道降级策略：
主渠道失败 → 备用渠道 → 兜底渠道
示例：
Slack (主) → Email (备) → In-App (兜底)
钉钉 (主) → SMS (备) → In-App (兜底)
Python
class ChannelFallback:
"""渠道降级管理器"""
FALLBACK_CHAINS = {
NotificationChannel.SLACK: [
NotificationChannel.EMAIL,
NotificationChannel.IN_APP,
],
NotificationChannel.DINGTALK: [
NotificationChannel.SMS,
NotificationChannel.IN_APP,
],
NotificationChannel.FEISHU: [
NotificationChannel.EMAIL,
NotificationChannel.IN_APP,
],
NotificationChannel.EMAIL: [
NotificationChannel.IN_APP,
],
NotificationChannel.SMS: [
NotificationChannel.IN_APP,
],
}
async def send_with_fallback(
self, primary_channel: NotificationChannel,
recipient: Recipient,
content: RenderedContent,
metadata: dict,
) -> DeliveryResult:
channels = [primary_channel] + self.FALLBACK_CHAINS.get(
primary_channel, [NotificationChannel.IN_APP]
)
for channel in channels:
adapter = self.adapters.get(channel)
if not adapter:
continue
if not await adapter.health_check():
continue
result = await adapter.send(
recipient, content, metadata
)
if result.success:
return result
self.logger.warning(
f"Channel {channel.value} failed for "
f"{recipient.user_id}: {result.error}, "
f"trying fallback"
)
return DeliveryResult(
success=False,
error="All channels failed",
)
#8. 监控与指标
#8.1 关键指标
Python
NOTIFICATION_METRICS = {
"notification_sent_total": Counter(
"notification_sent_total",
labels=["channel", "status", "template_id"],
),
"notification_latency_seconds": Histogram(
"notification_latency_seconds",
labels=["channel"],
buckets=[0.1, 0.5, 1, 2, 5, 10, 30],
),
"notification_rate_limited_total": Counter(
"notification_rate_limited_total",
labels=["channel", "user_id"],
),
"notification_fallback_total": Counter(
"notification_fallback_total",
labels=["primary_channel", "fallback_channel"],
),
"notification_digest_total": Counter(
"notification_digest_total",
labels=["user_id"],
),
}
#8.2 投递仪表板
Code
┌────────────────────────────────────────────────────────────┐
│ 通知投递仪表板 │
│ │
│ 过去 24 小时: │
│ ┌──────────┬───────┬────────┬────────┬─────────┐ │
│ │ 渠道 │ 发送 │ 成功 │ 失败 │ 成功率 │ │
│ ├──────────┼───────┼────────┼────────┼─────────┤ │
│ │ Email │ 1,234 │ 1,220 │ 14 │ 98.9% │ │
│ │ SMS │ 456 │ 450 │ 6 │ 98.7% │ │
│ │ DingTalk │ 2,345 │ 2,340 │ 5 │ 99.8% │ │
│ │ Feishu │ 1,890 │ 1,885 │ 5 │ 99.7% │ │
│ │ In-App │ 5,678 │ 5,678 │ 0 │ 100.0% │ │
│ │ Push │ 890 │ 845 │ 45 │ 94.9% │ │
│ └──────────┴───────┴────────┴────────┴─────────┘ │
│ │
│ 降级事件: 23 次 │ 限频触发: 156 次 │ 摘要合并: 89 条 │
└────────────────────────────────────────────────────────────┘
#Key Takeaways
- 渠道适配器模式：统一的
ChannelAdapter接口让新渠道接入只需实现 send/send_batch/health_check - 模板引擎：一套模板自动适配 HTML（Email）、Markdown（IM）、短文本（SMS/Push）等多种格式
- 偏好路由：根据用户配置和通知类别自动选择最合适的渠道
- 防骚扰：频率限制 + 安静时段 + 摘要合并，三重保护避免通知轰炸
- 降级链：主渠道不可用时自动切换备用渠道，In-App 作为最终兜底
- 完整追踪：每条通知从创建到投递的全生命周期有完整记录
#Next Article
下一篇 S5-17 用户自定义函数：多语言沙箱运行时设计 将详解 FunctionRuntime 如何支持 Python、TypeScript、Groovy、WASM 和 Kotlin 五种语言的安全隔离执行。
tags: notification, channel-adapter, template, rate-limit, fallback, digest, coomia-dip