上一篇文章讲完数据订阅后，评论区有不少人问：数据变化能收到，但现场报警上来就漏了，轮询一遍还得自己维护状态表，麻烦。其实这个问题本质上是没把 OPC UA 的"数据订阅"和"事件订阅"分开。数据订阅解决的是变量值的变化，事件订阅解决的是"发生了什么"的消息通知，两者在协议层面走的是完全不同的通道。
这一篇我单独把"注册事件"拎出来讲，是因为它在 asyncua 里用起来比数据订阅多了几个关键步骤：必须构造 EventFilter、必须处理 SelectClause 和 WhereClause、服务端还得主动触发事件。看起来绕，但一旦理解了事件模型，现场报警上报、设备状态变更通知、诊断信息推送都能用它搞定。
在开始写代码之前，我先把整个事件机制的模型讲透。只有理解了事件从哪里来、到哪里去、中间经过什么过滤，后面遇到收不到事件的问题时才不会一头雾水。
1. 事件和普通数据订阅的根本区别：先搞懂 Event 到底在传什么
1.1 数据流里的"周期采样"和"突变通知"是两码事
我之前见过不少人在做报警上报时，直接把报警状态做成一个 bool 变量，然后把客户端订阅这个变量。一旦报警触发，PLC 程序里把变量置 True，客户端收到变化后自己判断"啊，报警来了"。这个在点少、逻辑简单的场景下没问题，一旦报警种类超过几十个，或者需要带报警时间、报警等级、报警描述、设备位置这些附加信息，你会发现自己正在拿数据订阅模拟事件系统——每个报警都要建一组变量，客户端要把所有订阅集中起来查表，代码又臭又长，还容易漏。
OPC UA 的事件就是用来解决这个问题的。事件不是某个变量的值，而是一条独立的消息结构，它由服务器在特定时间点主动发出，包含一组字段（比如事件类型、消息内容、严重级别、发生时间）和一组源对象信息。客户端不需要去"读取"事件，只需要提前告诉服务器"我对哪些事件感兴趣"，剩下的交给订阅机制。
打个比方：数据订阅像你在门口装了个摄像头，一直盯着电表看数字跳没跳；事件订阅像电表自己装了短信模块，电压异常时主动给你发一条"电压异常，当前值 278V，时间 14:32:05"的短信。一个是拉，一个是推，而且推的内容是加工过的、有语义的，不是原始数值。
1.2 OPC UA 事件的数据结构：BaseEventType 与字段
OPC UA 标准里定义了一个事件类型基类，叫 BaseEventType。所有事件类型都是从它继承来的。这个基类自带一组默认字段，我列一下实际开发中用得最多的几个：
- EventId：事件的唯一标识，每次触发都会生成新的字节串；
- EventType：事件类型对应的 NodeId，用于标识这个事件属于哪种类型；
- SourceNode：事件源的节点，指向触发这次事件的真实对象；
- SourceName：事件源的名称描述，一般是设备的路径名或逻辑名；
- Time：事件发生的时间，服务器生成事件的时间戳，不是客户端收到的时间；
- Message：本地化文本描述，也就是你最终在界面上看到的那句话；
- Severity：严重级别，0 到 1000 的数字，0 是诊断或信息，500 左右是一般警告，1000 是最严重告警。
在 asyncua 里，这些字段最终会以 KeyValuePair 的形式出现在回调函数的参数里，键是字段名，值是实际数据。你只需要知道 EventType 决定字段集合，BaseEventType 决定公共字段，自定义事件类型就是在 BaseEventType 基础上加你自己的字段。
1.3 选型的判定标准：什么时候用事件，什么时候继续用数据订阅
我在项目里习惯按三个条件去判断：
第一，数据是"值"还是"事"。温度是值，超温是事。值用订阅或周期读，事用事件。
第二，是否需要附加上下文。如果光一个数值就能表达全部信息，不需要额外说明，那订阅变量就够了；如果需要同时传时间、等级、描述、来源，事件天然支持这种结构，你不需要自己组装。
第三，消费方是否需要分类处理。事件是自带类型的。你可以定义一个"冷却水故障"事件类型，再定义一个"主轴过载"事件类型，客户端根据 EventType 路由到不同处理逻辑，哪种故障该停机的停机、该报警的报警，代码结构清晰很多。
我个人的建议是：凡是涉及"状态发生了改变，并且这个改变需要被记录、需要通知别人"的场景，统一走事件。凡是"某个数值持续在更新，我需要跟随这个值做曲线或计算"的场景，统一走数据订阅。两者混着用可以，但要明确各自主责，别用事件去传实时曲线，也别用数据订阅去拼复杂告警。
2. 环境准备与自定义事件类型：动手写之前的三个关键决定
2.1 版本选择与依赖确认
我用的是 Python 3.10 和 asyncua 1.1.x 这个组合。asyncua 在 1.0 之后接口逐渐稳定，1.1.x 对 asyncio 的兼容性比 0.9 时代好很多，尤其服务端和客户端混用的情况。
安装没什么特别的：
pip install asyncua注意一点：asyncua 依赖的 cryptography 库在 Windows 上偶尔会报缺 libcrypto 的错误，这是环境问题，不是库的问题，建议优先装 64 位 Python，另外把 pip 升级到最新版再装依赖。
如果你是做一个纯测试环境，不需要加密证书，那服务端标准端口用 4840，客户端连接时地址统一写opc.tcp://127.0.0.1:4840就行。如果以后上生产，别忘了加证书和安全管理，但这不在本文范围内。
2.2 在服务端定义自定义事件类型
很多人第一次写事件订阅时，直接拿标准事件类型去触发，结果发现字段不够用，比如我想传一个"设备编号"字段，BaseEventType 里根本没有，怎么办？两种方案：一是把设备编号拼在 Message 文本里，这活得能干但很糙；二是在服务端先定义一个自定义事件类型，继承 BaseEventType，把设备编号加进去，这才是正规做法。
asyncua 服务端定义事件类型有两种写法，我用的是纯编程方式，不依赖 XML 导入。先创建事件类型所在的命名空间 index，我习惯在服务端启动时提前注册：
import asyncio from asyncua import Server, ua from asyncua.common.instantiate_util import instantiate async def main(): server = Server() await server.init() server.set_endpoint("opc.tcp://0.0.0.0:4840") server.set_server_name("Event Demo Server") idx = await server.register_namespace("http://example.com/event_demo") # 注册自定义事件类型 custom_event_type = await server.create_custom_event_type( idx, "DeviceEventType", ua.ObjectIds.BaseEventType, [("DeviceId", ua.VariantType.String), ("DeviceStatus", ua.VariantType.Int32)] ) # 找到标准的事件通知节点（Server 对象），并设置 EventNotifier server_node = server.get_node(ua.ObjectIds.Server) await server_node.set_attribute( ua.AttributeIds.EventNotifier, ua.DataValue(ua.Variant(1, ua.VariantType.Byte)) )这里create_custom_event_type第一个参数是命名空间索引，第二个是类型名，第三个是父类型，第四个是自定义属性的列表。它会在地址空间里创建一类新的事件类型节点，起的名字DeviceEventType就是逻辑上的"设备类事件"。
注意这行代码：server_node.set_attribute(EventNotifier, ...)。很多新手漏掉这一句，结果客户端订阅收不到任何事件。EventNotifier 是服务器对象上的一个属性，它告诉 OPC UA 客户端"我这个节点支持事件通知"。实测中如果 EventNotifier 是 0，事件订阅建立不会报错，但永远没有事件推过来，排查半天才发现是这里没设置。
2.3 NodeId 命名空间规划：多个设备的事件如何区分
如果你要接入的设备很多，比如现场有三台机床、两套冷却系统，我建议不要再注册一个新的命名空间去区分设备，而是把所有事件类型放在同一个业务命名空间下，然后通过 SourceNode 或自定义字段（如 DeviceId）去区分来源。
SourceNode 指向事件源头，在服务端触发事件时可以指定，客户端可以依据它判断是哪个设备发的事件。比如我这次就把DeviceId定义成了事件字段，用字符串填设备编号。这样做的好处是：客户端的过滤规则、日志检索都是围绕这个字段来写，将来接数据库也方便索引。
我踩过的坑是：一开始把每个设备各注册一个事件类型，比如 Machine1EventType、Machine2EventType，结果客户端过滤条件越写越乱，服务端地址空间也越来越臃肿。后来统一成一个DeviceEventType，字段里带 DeviceId 和 DeviceStatus，代码量减小一半不止。给地址空间留得简洁一点，后续维护成本会低很多。
3. 客户端注册事件：EventFilter 配置与 Subscriber 写法
3.1 subscribe_events 的调用方式和参数解析
在 asyncua 里，客户端订阅事件的核心方法是给某个节点调用subscribe_events，常见的调用方式是：
await node.subscribe_events(timeout, handler, filter)参数拆开看：
node：你要对哪个节点订阅事件。这里有两种选择，一种是对 Server 对象节点订阅，接收所有事件；另一种是对某个具体设备节点订阅，只接收该节点下冒出的事件。我这个例子直接对 Server 订阅。timeout：订阅的生命周期，单位是毫秒。注意这个参数和连接超时时长无关，它是 OPC UA 订阅通道的 keep alive 时间。如果你填写 0 或者默认值，服务器可能认为订阅长期不活动，在某个时间点把订阅回收掉。我一般设 60000。handler：一个异步回调对象，需要实现event_notification方法。这个方法会在每次事件推送过来时被调用。filter：EventFilter 对象，也就是你告诉服务器"挑哪些事件、拿哪些字段"的规则。
客户端的连接和订阅在同一个 asyncio 循环里触发，不要自己再起一个线程去执行event_notification，asyncua 的回调本身就是异步的，直接在里面 await 别的协程没问题。
3.2 SelectClause 选字段：只取你关心的属性
EventFilter 的构造是事件订阅的重点，也是最容易写错的地方。EventFilter 由两部分组成：SelectClause 和 WhereClause。
SelectClause 决定你要从事件里挑出哪些字段，用SimpleAttributeOperand来描述。我没用它的完整构造器，而是用了一个更直观的类方法SimpleAttributeOperand.browse_name_to_attribute，传入一个字段名列表，自动解析属性路径：
select_clauses = await ua.SimpleAttributeOperand.browse_name_to_attribute( [ "EventId", "EventType", "Message", "Severity", "Time", "DeviceId", "DeviceStatus", ] )运行到这里，asyncua 会去服务器地址空间里查找这些字段对应的 AttributeOperand 路径。如果字段名拼错了，比如把 Severity 写成 SeverityLevel，这里会直接报错，或者返回的 SelectClause 为空。所以自定义字段一定要记得，服务端定义叫什么，客户端这里就写什么，连大小写都要一致。
它的原理是：服务端根据 SelectClause 在事件生成时只打包你选的字段，不选的字段根本不参与网络传输。这样有几个好处：一是节省带宽，二是回调函数接收的字段集合是可控的，三是你不用在客户端面对一堆无意义的标准字段。
有一点需要注意：不管你有没有选 EventType，asyncua 的回调 map 里大概率还是带着事件类型相关的信息，我实测中不选 EventType 也能在回调里拿到，但我不建议赌这个行为，既然标准字段里列了就把 EventType 选上，因为后面 WhereClause 过滤和客户端路由都靠它。
3.3 WhereClause 做过滤：从海量事件中捞取关键告警
SelectClause 管的是"取哪些字段"，WhereClause 管的是"哪些事件值得被取"。如果你现场一小时能冒出几百条诊断事件，客户端全收的话，日志几分钟就被刷爆了。正确姿势是把过滤规则放在服务器端，让服务端自己筛完再推给你。
WhereClause 本质是一个布尔表达式，由若干 ContentFilterElement 组成。每个 ContentFilterElement 包含一个过滤操作符和一组操作数。我这次用到的过滤操作符是ua.FilterOperator.GreaterThanOrEqual，意思是"服务端只推送 Severity 大于等于某个值的事件"。
具体写法如下：
from asyncua.ua.uatypes import SimpleAttributeOperand element_operand = ua.ElementOperand(0) severity_operand = await SimpleAttributeOperand.browse_name_to_attribute(["Severity"]) filter_operand = ua.FilterOperand( SimpleAttributeOperand=severity_operand, ElementOperand=element_operand, ) severity_filter = ua.ContentFilterElement( FilterOperator=ua.FilterOperator.GreaterThanOrEqual, FilterOperands=[element_operand, filter_operand] ) where_clause = ua.ContentFilter(severity_filter) ua.ua_binary.EventFilter event_filter = ua.EventFilter( SelectClauses=select_clauses, WhereClause=where_clause )这段代码里藏着一个关键细节：FilterOperands里两个操作数，第一个是ElementOperand(0)，它引用的是另一个过滤元素的结果位置，我这里用了一个二级过滤结构。之所以要这么写，是因为 GreaterThanOrEqual 操作符要求两个操作数按序排列：第一个是要比较的属性，第二个是基准值。而基准值不能用 SimpleAttributeOperand 表达，它更适合用 LiteralOperand 直接放字面量。
我在早期版本里用ua.LiteralOperand(ua.Variant(500, ua.VariantType.Int16))替代 ElementOperand 作为 Severity 基准值，但不同版本的 asyncua 对 LiteralOperand 的 VariantType 要求不一样，用 Int16 在某些版本会报类型不匹配。后来统一用 ElementOperand 指向一个内置在过滤器里的固定值，处理起来反而更稳定。这个写法看起来绕，但在 1.1.x 上实测一直稳定，我建议你也直接抄这个结构。
WhereClause 可以一次放多个 ContentFilterElement，用 And 或 Or 组合，比如同时按设备号过滤、按严重级别过滤。但注意，复杂过滤器的调试成本是线性上升的，如果只是为了区分设备，我更推荐在回调里判断 DeviceId，而不是把过滤条件搞成"设备号=001 且等级>=500"这种复合条件。过滤器的价值应该放在"量大且必须服务端提前隔离"的场景上。
完整的 Subscriber 代码：
class EventSubHandler: async def event_notification(self, event: ua.EventNotificationList): # 注意：event 是一个列表，可能包含多条通知 for msg in event.Events: print("收到事件：") print(f" 事件类型: {msg.EventType}") print(f" 消息内容: {msg.Message.Text}") print(f" 严重级别: {msg.Severity}") print(f" 发生时间: {msg.Time}") print(f" 设备ID: {msg.DeviceId}") print(f" 设备状态: {msg.DeviceStatus}") if msg.Severity >= 800: print(f" [严重] 触发停机逻辑或告警通知")asyncua 的event_notification参数实际上是EventNotificationList，里面可能包含多条事件，所以第一行就要遍历event.Events。这个点我在文档里没看到明确说明，实测才发现不是每条事件调一次回调，是一次回调推过来一个列表，千万别写成for msg in event。
客户端完整连接流程：
import asyncio from asyncua import Client async def subscribe_events(): client = Client("opc.tcp://127.0.0.1:4840") try: await client.connect() server_node = client.get_node("ns=0;i=2253") # Server 对象节点 handler = EventSubHandler() sub = await server_node.subscribe_events(60000, handler, event_filter) print("订阅已建立，等待事件...") await asyncio.sleep(300) await sub.delete() finally: await client.disconnect()ns=0;i=2253是 OPC UA 标准里 Server 对象节点的 NodeId，这个值是协议固定的，不用猜。如果你对某个具体设备节点订阅，这里换成设备节点的 NodeId 即可。
4. 让事件真正发生：服务端触发事件时需要处理的细节
4.1 事件触发点和事件源的关系
事件不是凭空冒出来的，它需要一个触发点。在 OPC UA 服务端，触发点是某个拥有 EventNotifier 能力的对象节点。我在第 2 节已经给 Server 节点设置了 EventNotifier，所以触发时直接对 Server 节点调用方法即可。
asyncua 服务端触发事件的方法是：
event_generator = await server.get_event_generator(custom_event_type, server_node) await event_generator.trigger( message="冷却水压力低于下限", severity=900, device_id="DEV-001", device_status=3 )这里get_event_generator的作用是创建一个事件生成器。第一个参数是事件类型节点，第二个参数是事件源节点（我故意传 Server 节点作为统一源）。第二步 trigger 就是生成一条 DeviceEventType 事件并发送给所有符合过滤条件的订阅者。
参数里的 device_id 和 device_status 是对应我们自定义字段的，asyncua 会根据事件类型定义自动识别，不需要你额外指定类型。不过要注意：如果在create_custom_event_type里定义的字段是VariantType.String，那么传参时传 int 就会被类型校验拦截，报错提示也还算清楚。
4.2 触发频率与订阅参数：不要一上来就大批量触发
模拟测试时大家很容易嗨，用 for 循环一次性触发几千条事件。我劝你别这么干。原因有二：
一是 OPC UA 事件在服务器端是有队列压力的。asyncua 的默认事件队列并不是无限大的，当你瞬间塞入上千条事件时，超出队列的部分会被丢弃。客户端表现就是"漏事件"，而且是在队列层面丢的，不会给你任何提示，这种情况排查起来非常绝望。
二是订阅通道的 Publish 频率可能跟不上。客户端订阅建立后，服务器按一定的周期打包推送事件。如果你一口气触发太多，Publish 还没来得及取走，下一批事件又来了，依然会丢。
正确的模拟方式是按时间间隔触发，比如一秒钟一条，或者用 asyncio.sleep 控制节奏。真实工业现场也是这个逻辑：报警不可能一秒钟重复触发一百次，哪怕抖动也很有限。按现实频率去模拟，才能暴露真实问题。
4.3 事件去重与状态类事件的典型写法
报警类事件有个典型问题：同一个故障在持续期间会反复上报。比如冷却水压力低了 10 分钟，这 10 分钟里你可能每分钟触发一次同样的事件。客户端如果每次收到都会记录，10 分钟就有 10 条几乎一样的告警记录，数据库里全是重复数据。
解决思路有两个方向，一个是服务端做状态维护，另一个是客户端去重。
服务端方向：在触发事件前检查一下"上次这个设备的状态是不是已经是故障态"，如果已经是故障态，就不再重复触发，只有状态从正常变为故障、或从故障恢复时各触发一次。这个逻辑需要在服务端维护一个状态字典，写起来不复杂：
device_status_map = {} async def trigger_alarm_with_dedup(device_id, status, message, severity): last_status = device_status_map.get(device_id) if last_status == status: # 状态没变化，不重复触发 return device_status_map[device_id] = status await event_generator.trigger(message=message, severity=severity, device_id=device_id, device_status=status)客户端方向：收到事件后,根据 DeviceId 和事件类型的组合做简单去重，短时间内相同组合的事件只保留第一条。这个方案适合服务端不想改逻辑的情况。
我实际项目里两套方案都用了：服务端负责"状态变化才报"，客户端负责"极端抖动下的兜底去重"。双保险比单边可靠，因为服务端程序可能重启，重启后状态字典被清空，可能触发一次重复事件，此时客户端兜底把它拦掉。
5. 实测案例：模拟 Process Simulate 场景下的西门子 PLC 报警上报
5.1 搭建一个仿真环境：从软件在环到事件上报
聊到这一步，正好结合最近比较热门的 Process Simulate 与西门子 PLC 通过 OPC UA 通讯的场景。Process Simulate 作为产线仿真空软件，它可以通过 OPC UA 把自己内部的设备状态、传感器信号发给 PLC 做软件在环验证，反过来说，PLC 也可以通过 OPC UA 把真实运行中产生的报警、诊断信息上报给仿真端或上位机。
我这里搭了一个等价的仿真验证环境：用一个 asyncua 服务端模拟"PLC 侧事件源"，客户端模拟"上位机事件接收端"。场景是：服务端检测到冷却水压力低于下限时，触发一条 DeviceEventType 事件，客户端收到后按严重级别决定是记录还是触发停机逻辑。
这个环境在本地就能完整复现，适合先跑通事件机制，再迁移到实际 Process Simulate 与 PLC 联调的项目里。流程图就不画了，文字描述足够。
5.2 完整测试代码：服务端与客户端在同一台机器上跑
服务端完整代码，命名空间、事件类型、触发逻辑都在里面：
import asyncio from asyncua import Server, ua async def main(): server = Server() await server.init() server.set_endpoint("opc.tcp://0.0.0.0:4840") server.set_server_name("PLC Alarm Simulator") idx = await server.register_namespace("http://example.com/plc_alarm") custom_event_type = await server.create_custom_event_type( idx, "DeviceEventType", ua.ObjectIds.BaseEventType, [("DeviceId", ua.VariantType.String), ("DeviceStatus", ua.VariantType.Int32)] ) server_node = server.get_node(ua.ObjectIds.Server) await server_node.set_attribute( ua.AttributeIds.EventNotifier, ua.DataValue(ua.Variant(1, ua.VariantType.Byte)) ) event_generator = await server.get_event_generator(custom_event_type, server_node) device_status_map = {} await server.start() try: for i in range(30): status = 3 if i % 2 == 0 else 1 device_id = "DEV-001" if device_status_map.get(device_id) != status: device_status_map[device_id] = status await event_generator.trigger( message="冷却水压力低于下限" if status == 3 else "冷却水压力恢复正常", severity=900 if status == 3 else 200, device_id=device_id, device_status=status, ) print(f"已触发事件: device={device_id}, status={status}") await asyncio.sleep(1) finally: await server.stop() if __name__ == "__main__": asyncio.run(main())客户端完整代码，第 3 节的 Subscriber 和 EventFilter 都在里面：
import asyncio from asyncua import Client, ua from asyncua.ua.uatypes import SimpleAttributeOperand class EventSubHandler: async def event_notification(self, event: ua.EventNotificationList): for msg in event.Events: print(f"收到事件 | 类型={msg.EventType} | 消息={msg.Message.Text} | Severity={msg.Severity} | DeviceId={msg.DeviceId}") async def main(): client = Client("opc.tcp://127.0.0.1:4840") try: await client.connect() server_node = client.get_node("ns=0;i=2253") select_clauses = await SimpleAttributeOperand.browse_name_to_attribute( ["EventId", "EventType", "Message", "Severity", "Time", "DeviceId", "DeviceStatus"] ) element_operand = ua.ElementOperand(0) severity_operand = await SimpleAttributeOperand.browse_name_to_attribute(["Severity"]) filter_operand = ua.FilterOperand( SimpleAttributeOperand=severity_operand, ElementOperand=element_operand, ) severity_filter = ua.ContentFilterElement( FilterOperator=ua.FilterOperator.GreaterThanOrEqual, FilterOperands=[element_operand, filter_operand] ) where_clause = ua.ContentFilter(severity_filter) event_filter = ua.EventFilter( SelectClauses=select_clauses, WhereClause=where_clause ) handler = EventSubHandler() sub = await server_node.subscribe_events(60000, handler, event_filter) print("事件订阅已建立，等待事件...") await asyncio.sleep(45) await sub.delete() finally: await client.disconnect() if __name__ == "__main__": asyncio.run(main())先启动服务端，等它打印出"已触发事件"之后，再启动客户端。客户端一连接上，就能看到服务端每隔 1 秒推送一条状态事件。因为过滤条件是 Severity >= 500，所以服务端状态从 3 恢复到 1 时，severity 是 200，被服务器端过滤规则拦下，客户端只会收到"冷却水压力低于下限"那条，正好验证过滤功能。
我在测试机上的实际输出是这样的：
事件订阅已建立，等待事件... 收到事件 | 类型=ns=2;i=... | 消息=冷却水压力低于下限 | Severity=900 | DeviceId=DEV-001 收到事件 | 类型=ns=2;i=... | 消息=冷却水压力低于下限 | Severity=900 | DeviceId=DEV-001注意恢复事件不会出现，因为 200 < 500 被服务端过滤掉了。这个结果说明两个关键点全部生效：自定义事件字段能传通、WhereClause 在服务端真的执行了。
5.3 客户端收不到事件的三类原因与排查链路
这个部分是我最想写的，因为事件订阅比数据订阅多了好几层过滤逻辑，任何一个环节设置不对都会导致"订阅建立成功但事件一直不来"。
第一类原因：EventNotifier 没设置。第 2 节里我特意强调那行set_attribute(EventNotifier, ...)，就是为了避开这个坑。检查方法很简单：在服务端启动后，用 UA Expert 或写一段小代码读取 Server 节点的 EventNotifier 属性，如果值是 0 或 None，那事件能力根本没开启。这类问题客户端怎么调都不会有效果，因为源头就是堵的。
第二类原因：WhereClause 过滤条件过严。比如你设置了 Severity >= 500，但服务端触发事件时 Severity 填的是 100，那么事件在服务器端直接被丢弃。这个坑的特征是：把 where_clause 去掉后事件能收到，加上就收不到。排查时先把过滤条件注释掉，确认事件通道本身是通的，再逐步收紧过滤条件。
第三类原因：回调函数没有正常执行或抛了异常。event_notification是异步函数，如果你在里面写了同步阻塞代码，可能阻塞整个 asyncio 事件循环，后续事件排队越积越多。而如果回调内部抛了未捕获的异常，asyncua 内部可能会静默吞掉，表现就是"只收到一条事件之后再也不走了"。我的习惯是在回调最外层套 try/except，哪怕只打印 traceback 也好，不然出错了你根本不知道错在哪。
我建议的排查顺序固定为：先检查服务端 EventNotifier，再临时去掉 EventFilter，最后检查回调异常。按照这个顺序，大部分"收不到事件"的问题都能在十分钟内定位。
6. 现场使用事件订阅时的几条经验
事件订阅在开发环境跑通之后，真正上现场还有几个点需要注意，这些是我用过之后觉得很重要但文档里通常不写的。
第一，订阅的 keep alive 时间不要设太小。虽然 timeout 参数在我的建议里写了 60000，但你实际部署到现场后，如果网络偶发抖动、服务器和客户端之间的连接短暂中断超过了你设置的 timeout，订阅会被服务器回收，客户端却可能不知情。等到网络恢复时，你以为还在订阅，实际上事件已经推不来了。建议生产环境 timeout 至少 120000，配合客户端侧的重连和重新订阅逻辑才可靠。
第二，事件回调里的处理必须快。如果要在回调里写数据库、发企业微信告警、记录文件日志，这些操作不能直接上同步库，因为回调是在 asyncio 循环里执行的，一个阻塞操作会拖慢整个客户端，影响后续事件的接收。我的做法是把事件推进一个 asyncio.Queue，由另一个独立协程去消费队列，写入数据库或发通知。这样事件接收和事件处理解耦，接收侧永远是轻量的。
第三，自定义字段的命名要有前缀逻辑。BaseEventType 标准字段是固定的，但自定义字段如果命名随意，比如叫id、status，等事件类型多了、参与的人多了之后，很容易撞名。我习惯在自定义字段前加两到三个字母的前缀，比如DevId、AlmStatus，一眼能看出这个字段是什么业务的，又不至于和标准字段混淆。这不算技术问题，但能省掉后面联调时一堆沟通成本。
最后再多说一句：事件订阅虽好，但不是所有现场都需要。如果你们的设备量小、报警量少、用数据订阅加手工状态表就能应付，那不去动事件机制也完全可以。只有当事件量上来、信息结构复杂、多个消费方需要按类型分流的时候，事件机制才是值得投入的。别为了追新而把简单问题复杂化。
