如何使用Rx.NET根据历史消息处理规则调整消息流的处理方式？

这问题我太熟了！用Rx.NET实现这种能根据历史处理结果动态调整逻辑的智能消息处理器，核心就是利用Rx的状态管理能力和操作符组合来搞定。我结合自动售货机的场景给你拆解下具体方案：

你的需求本质是带状态的消息流处理：处理器需要记住之前的处理结果（也就是当前系统状态），然后根据这个状态决定如何处理新消息。Rx.NET里有两个核心工具能帮我们实现这个：

• Scan操作符：可以在流中累积状态，每收到一条消息就结合当前状态生成新状态，同时输出这个状态变化。
• BehaviorSubject/ReplaySubject：用来保存当前的系统状态，让处理逻辑能随时感知状态变化，甚至基于状态切换不同的处理分支。

这种方式适合状态逻辑相对简单的场景，所有状态转换逻辑集中在一个地方，代码直观。

1. 定义状态模型和消息结构 #

首先我们需要明确系统的状态和消息格式：

// 自动售货机的状态枚举
public enum VendingMachineState
{
    Idle,
    AcceptingPayment,
    DispensingItem,
    OutOfStock
}

// 具体的状态数据模型
public class MachineState
{
    public VendingMachineState CurrentState { get; set; } = VendingMachineState.Idle;
    public decimal TotalInserted { get; set; } // 已投金额
    public string SelectedItem { get; set; } // 选中的商品ID
}

// 消息结构（适配你的IObservable<Message>）
public class Message
{
    public string Command { get; set; } // 比如"P1"=选商品1，"C0.5"=投0.5元
    public object Payload { get; set; }
}

2. 用Scan维护状态并处理消息 #

Scan会帮我们把每个消息和当前状态结合，生成新状态，我们只需要在里面写状态转换的逻辑：

// 初始化初始状态
var initialState = new MachineState();

// 模拟你的消息流（实际是从消息队列/服务总线来的IObservable<Message>）
var messageStream = new Subject<Message>();

// 生成带状态的消息处理流
var statefulProcessingStream = messageStream
    .Scan(initialState, (currentState, incomingMsg) =>
    {
        // 根据当前状态和消息，计算新状态
        switch (currentState.CurrentState)
        {
            case VendingMachineState.Idle:
                // 空闲状态下只处理选品指令
                if (incomingMsg.Command.StartsWith("P"))
                {
                    var itemId = incomingMsg.Command.Substring(1);
                    bool isInStock = CheckItemStock(itemId); // 假设的库存检查方法
                    
                    return isInStock 
                        ? new MachineState
                          {
                              CurrentState = VendingMachineState.AcceptingPayment,
                              SelectedItem = itemId,
                              TotalInserted = currentState.TotalInserted
                          }
                        : new MachineState
                          {
                              CurrentState = VendingMachineState.OutOfStock,
                              SelectedItem = itemId,
                              TotalInserted = currentState.TotalInserted
                          };
                }
                break;

            case VendingMachineState.AcceptingPayment:
                // 收款状态下只处理投币指令
                if (incomingMsg.Command.StartsWith("C"))
                {
                    decimal insertedAmount = decimal.Parse(incomingMsg.Command.Substring(1));
                    decimal newTotal = currentState.TotalInserted + insertedAmount;
                    decimal itemPrice = GetItemPrice(currentState.SelectedItem); // 假设的商品价格方法
                    
                    return newTotal >= itemPrice
                        ? new MachineState
                          {
                              CurrentState = VendingMachineState.DispensingItem,
                              SelectedItem = currentState.SelectedItem,
                              TotalInserted = newTotal
                          }
                        : new MachineState
                          {
                              CurrentState = VendingMachineState.AcceptingPayment,
                              SelectedItem = currentState.SelectedItem,
                              TotalInserted = newTotal
                          };
                }
                break;

            case VendingMachineState.DispensingItem:
                // 出货状态下处理完成指令
                if (incomingMsg.Command == "DispenseComplete")
                {
                    return new MachineState
                    {
                        CurrentState = VendingMachineState.Idle,
                        SelectedItem = null,
                        TotalInserted = 0
                    };
                }
                break;

            case VendingMachineState.OutOfStock:
                // 缺货状态下处理补货指令
                if (incomingMsg.Command == "Restock")
                {
                    return new MachineState
                    {
                        CurrentState = VendingMachineState.Idle,
                        SelectedItem = null,
                        TotalInserted = 0
                    };
                }
                break;
        }
        // 如果没有匹配的处理逻辑，保持当前状态
        return currentState;
    });

// 订阅处理结果，执行实际的业务动作（比如通知硬件、更新UI）
statefulProcessingStream.Subscribe(updatedState =>
{
    Console.WriteLine($"当前状态: {updatedState.CurrentState} | 已投金额: {updatedState.TotalInserted} | 选中商品: {updatedState.SelectedItem}");
    
    switch (updatedState.CurrentState)
    {
        case VendingMachineState.AcceptingPayment:
            decimal remaining = GetItemPrice(updatedState.SelectedItem) - updatedState.TotalInserted;
            Console.WriteLine($"还需投币: {remaining:C}");
            break;
        case VendingMachineState.DispensingItem:
            Console.WriteLine($"正在出货: {updatedState.SelectedItem}");
            // 模拟出货完成，发送完成消息
            messageStream.OnNext(new Message { Command = "DispenseComplete" });
            break;
        case VendingMachineState.OutOfStock:
            Console.WriteLine($"商品{updatedState.SelectedItem}缺货，请补货");
            break;
    }
});

// 模拟发送测试消息
messageStream.OnNext(new Message { Command = "P1" });
messageStream.OnNext(new Message { Command = "C0.5" });
messageStream.OnNext(new Message { Command = "C0.3" });

如果你的状态逻辑很复杂，每个状态的处理规则差异很大，推荐用这种方式：把每个状态对应的处理逻辑拆成独立的子流，然后根据当前状态动态切换到对应的子流。

// 用BehaviorSubject保存当前状态，方便随时感知状态变化
var currentStateSubject = new BehaviorSubject<MachineState>(initialState);

// 基于当前状态动态切换处理流
var dynamicProcessingStream = currentStateSubject
    .Select(currentState =>
    {
        // 根据当前状态返回对应的消息处理子流
        switch (currentState.CurrentState)
        {
            case VendingMachineState.Idle:
                // 空闲状态：只处理选品消息，生成新状态
                return messageStream
                    .Where(msg => msg.Command.StartsWith("P"))
                    .Select(msg => ProcessSelectItemCommand(msg, currentState));
            case VendingMachineState.AcceptingPayment:
                // 收款状态：只处理投币消息，生成新状态
                return messageStream
                    .Where(msg => msg.Command.StartsWith("C"))
                    .Select(msg => ProcessPaymentCommand(msg, currentState));
            case VendingMachineState.DispensingItem:
                // 出货状态：只处理完成消息，生成新状态
                return messageStream
                    .Where(msg => msg.Command == "DispenseComplete")
                    .Select(msg => ProcessDispenseCompleteCommand(msg, currentState));
            case VendingMachineState.OutOfStock:
                // 缺货状态：只处理补货消息，生成新状态
                return messageStream
                    .Where(msg => msg.Command == "Restock")
                    .Select(msg => ProcessRestockCommand(msg, currentState));
            default:
                return Observable.Empty<MachineState>();
        }
    })
    .Switch(); // 关键操作符：自动切换到最新状态对应的处理流

// 订阅处理结果，更新当前状态
dynamicProcessingStream.Subscribe(newState => currentStateSubject.OnNext(newState));

// 独立的处理函数，每个函数只负责一个状态的逻辑
private MachineState ProcessSelectItemCommand(Message msg, MachineState currentState)
{
    var itemId = msg.Command.Substring(1);
    bool isInStock = CheckItemStock(itemId);
    return isInStock
        ? new MachineState { CurrentState = VendingMachineState.AcceptingPayment, SelectedItem = itemId }
        : new MachineState { CurrentState = VendingMachineState.OutOfStock, SelectedItem = itemId };
}

private MachineState ProcessPaymentCommand(Message msg, MachineState currentState)
{
    decimal amount = decimal.Parse(msg.Command.Substring(1));
    decimal newTotal = currentState.TotalInserted + amount;
    decimal price = GetItemPrice(currentState.SelectedItem);
    return newTotal >= price
        ? new MachineState { CurrentState = VendingMachineState.DispensingItem, TotalInserted = newTotal, SelectedItem = currentState.SelectedItem }
        : new MachineState { CurrentState = VendingMachineState.AcceptingPayment, TotalInserted = newTotal, SelectedItem = currentState.SelectedItem };
}

// 其他处理函数同理...

• 状态不可变性：上面的示例中我们都是生成新的MachineState对象，而不是修改原对象，这能避免Rx流中的并发问题，保证状态的一致性。
• 错误处理：别忘了给流添加Catch/Retry操作符处理异常，避免单个消息处理失败导致整个流终止。
• 资源清理：如果你的处理逻辑涉及到外部资源（比如硬件连接），要在流结束时用Finally操作符清理资源。

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