Reactive Extensions模式匹配操作符：如何检测特定事件序列？

要实现这种基于事件类型序列的匹配通知，最直接高效的方式是用**有限状态机（FSM）**来建模匹配规则，每个匹配步骤对应状态机的一个节点，事件流中的每个事件驱动状态转移，当到达最终匹配状态时触发通知。

核心实现思路 #

1. 状态建模：把链式匹配规则拆解成一系列状态节点，比如初始状态、匹配MessageHandled后的状态、匹配TransactionCommitted后的状态、允许任意事件的状态，以及最终匹配成功的状态。
2. 转移规则：为每个状态定义触发转移的条件——比如WhenType<T>就是"收到T类型事件时转移到下一个状态"；ThenAnyNumberOfAnyType是"收到任意事件都保持当前状态"。
3. 事件驱动：每个事件到来时，根据当前状态检查是否符合转移条件，更新状态；一旦进入最终成功状态，立即触发通知。
4. 订阅管理：提供订阅/注销接口，让匹配器可以绑定到事件流，避免内存泄漏。

代码示例（C#） #

下面是一个贴合链式API风格的可复用实现：

匹配器与订阅器实现 #

public class EventMatcher
{
    private State _currentState;
    private readonly State _initialState = new("Initial");
    private readonly State _successState = new("Success");

    public EventMatcher()
    {
        _currentState = _initialState;
    }

    // 匹配指定类型事件（初始步骤用WhenType，后续用ThenType）
    public EventMatcher WhenType<T>()
    {
        var nextState = new State($"Matched_{typeof(T).Name}");
        _currentState.AddTransition(typeof(T), nextState);
        _currentState = nextState;
        return this;
    }

    public EventMatcher ThenType<T>() => WhenType<T>();

    // 允许任意数量的任意事件
    public EventMatcher ThenAnyNumberOfAnyType()
    {
        // 当前状态添加自转移规则，任意事件都留在当前状态
        _currentState.AddTransition(null, _currentState);
        return this;
    }

    // 完成规则定义，返回订阅器
    public EventSubscriber Subscribe(Action onMatch)
    {
        // 最后一步的状态触发后进入成功状态
        _currentState.AddTransition(null, _successState);
        return new EventSubscriber(this, onMatch);
    }

    // 内部状态类，管理状态转移规则
    private class State
    {
        public string Name { get; }
        private readonly Dictionary<Type?, State> _transitions = new();

        public State(string name) => Name = name;

        public void AddTransition(Type? eventType, State nextState)
        {
            _transitions[eventType] = nextState;
        }

        public State? GetNextState(Type eventType)
        {
            // 优先匹配指定类型，再匹配任意类型（null代表任意）
            if (_transitions.TryGetValue(eventType, out var state))
                return state;
            return _transitions.TryGetValue(null, out state) ? state : null;
        }
    }

    // 订阅器类，处理事件流的输入
    public class EventSubscriber
    {
        private readonly EventMatcher _matcher;
        private readonly Action _onMatch;
        private State _currentState;
        private readonly object _lockObj = new(); // 线程安全锁

        public EventSubscriber(EventMatcher matcher, Action onMatch)
        {
            _matcher = matcher;
            _onMatch = onMatch;
            _currentState = matcher._initialState;
        }

        // 处理到来的事件
        public void HandleEvent(object evnt)
        {
            lock (_lockObj)
            {
                var nextState = _currentState.GetNextState(evnt.GetType());
                if (nextState == null)
                {
                    // 匹配中断，重置到初始状态
                    _currentState = _matcher._initialState;
                    return;
                }

                _currentState = nextState;
                // 检查是否到达成功状态
                if (_currentState == _matcher._successState)
                {
                    _onMatch.Invoke();
                    // 重置状态以支持重复匹配，若只需单次匹配可在此调用Unsubscribe
                    _currentState = _matcher._initialState;
                }
            }
        }

        // 注销订阅，需与事件流的解绑逻辑配合
        public void Unsubscribe()
        {
            // 示例：如果事件流是委托触发，移除HandleEvent
            // eventStream.OnEvent -= HandleEvent;
        }
    }
}

使用示例 #

// 构建你的匹配规则
var subscriber = new EventMatcher()
    .WhenType<MessageHandled>()
    .ThenType<TransactionCommitted>()
    .ThenAnyNumberOfAnyType()
    .ThenType<QueueStopped>()
    .Subscribe(() => Console.WriteLine("触发目标事件序列通知！"));

// 绑定到事件流（假设eventStream有OnEvent委托）
eventStream.OnEvent += subscriber.HandleEvent;

// 不再需要时注销
// subscriber.Unsubscribe();

扩展优化建议 #

• 自定义条件匹配：除了按类型匹配，可扩展支持基于事件属性的条件（比如When<T>(e => e.Status == Success)），只需把Type参数换成Predicate<object>即可。
• 单次/重复匹配：在订阅时可添加参数，指定匹配成功后是否自动注销或重置状态。
• 调试日志：给状态机添加日志输出，记录每次状态转移和事件处理情况，方便排查匹配异常。
• 多事件流支持：让匹配器可以同时绑定到多个事件流，或者支持规则复用。

内容的提问来源于stack exchange，提问作者asgerhallas