ReactiveX适配性咨询：按指定顺序匹配同设备消息的业务实现

Great question—Rx.NET is perfectly suited for this sequential, per-device message matching scenario, and it handles all your edge cases (like device resets interrupting incomplete flows) gracefully. Let’s walk through how to implement this step by step.

Core Approach #

Your requirements boil down to three key tasks:

1. Persist every incoming message to the database first, no exceptions.
2. Group messages by device ID since only same-device A→B→C→D sequences count.
3. Track per-device flow state: Start on A, progress through B→C→D, and reset the entire flow immediately if a new A arrives before D completes.

Full Rx.NET Code Example #

First, define a simple model for your incoming messages:

public class DeviceMessage
{
    public string TypeIdentifier { get; set; } // "A", "B", "C", "D", etc.
    public string DeviceId { get; set; }
    // Add any other fields (timestamp, payload, etc.) as needed
}

Now, here’s the Rx pipeline to handle your workflow:

using System.Reactive.Linq;
using System.Reactive.Subjects;

// Assume this is your incoming message stream (could be from a queue, socket, etc.)
var incomingMessages = new Subject<DeviceMessage>();

// Step 1: Persist ALL messages to the database first (use DoAsync for async DB calls)
var persistedMessages = incomingMessages
    .Do(msg => SaveMessageToDatabase(msg))
    .Publish()
    .RefCount(); // Multicast to avoid duplicate DB writes if we have multiple subscribers

// Step 2: Group messages by device ID to isolate per-device flows
var perDeviceMessageStreams = persistedMessages
    .GroupBy(msg => msg.DeviceId);

// Step 3: Process each device's message stream to track A→B→C→D flows
var completedDeviceFlows = perDeviceMessageStreams
    .SelectMany(deviceGroup =>
        deviceGroup
            // Use Scan to track the current state of the device's flow
            .Scan(new { CurrentStage = "Idle", FlowReset = false }, (currentState, msg) =>
            {
                // If a new A arrives, reset the flow immediately (ignore any incomplete steps)
                if (msg.TypeIdentifier == "A")
                {
                    return new { CurrentStage = "WaitingForB", FlowReset = true };
                }

                // Progress the flow only if we receive the expected next message
                return currentState.CurrentStage switch
                {
                    "WaitingForB" when msg.TypeIdentifier == "B" => new { CurrentStage = "WaitingForC", FlowReset = false },
                    "WaitingForC" when msg.TypeIdentifier == "C" => new { CurrentStage = "WaitingForD", FlowReset = false },
                    "WaitingForD" when msg.TypeIdentifier == "D" => new { CurrentStage = "FlowCompleted", FlowReset = false },
                    // Ignore any messages that don't match the current expected stage
                    _ => currentState
                };
            })
            // Filter for events we care about: completed flows or resets
            .Where(state => state.CurrentStage == "FlowCompleted" || state.FlowReset)
            // Map to a human-readable result (or pass the full state for further processing)
            .Select(state => 
                state.CurrentStage == "FlowCompleted" 
                    ? $"Device {deviceGroup.Key}: Successfully completed A→B→C→D flow"
                    : $"Device {deviceGroup.Key}: Reset flow due to new A message"
            )
    );

// Subscribe to the completed flows to trigger your post-processing logic
completedDeviceFlows.Subscribe(
    result => Console.WriteLine(result),
    error => Console.WriteLine($"Flow processing error: {error.Message}")
);

// Helper method for database persistence (replace with your actual DB logic)
void SaveMessageToDatabase(DeviceMessage msg)
{
    // Your DB insert logic here: e.g., _dbContext.Messages.Add(msg); _dbContext.SaveChanges();
    Console.WriteLine($"Saved message {msg.TypeIdentifier}{msg.DeviceId} to database");
}

Key Logic Breakdown #

• Do/DoAsync: Ensures every incoming message is saved to the database before any flow processing happens. The Publish().RefCount() prevents duplicate DB writes if you add more subscribers later.
• GroupBy: Isolates each device’s message stream so flows from different devices don’t interfere with each other.
• Scan: This is the heart of the flow tracking—it maintains the current stage of the device’s workflow (Idle → WaitingForB → WaitingForC → WaitingForD → FlowCompleted) and handles resets when a new A arrives.
• State Filtering: The Where clause lets you react only to meaningful events (completed flows or resets), so you can trigger your post-analysis or cleanup logic accordingly.

Handling Edge Cases #

Duplicate Messages #

Since devices resend messages until confirmed, Rx will automatically ignore duplicates that don’t match the current flow stage. For example, if a device resends B after we’ve already moved to WaitingForC, the Scan operator will just keep the current state instead of regressing.

Timeout for Stuck Flows #

If you want to automatically reset flows that get stuck (e.g., no C arrives after B), add a Timeout operator to the per-device stream:

deviceGroup
    .Timeout(TimeSpan.FromMinutes(10), Observable.Return(new DeviceMessage { TypeIdentifier = "Timeout", DeviceId = deviceGroup.Key }))
    .Scan(...)
    // Update the Scan logic to handle the "Timeout" message by resetting the flow

Post-Processing on Completed Flows #

Instead of outputting a string, you can map the completed state to a custom object containing all the messages in the flow (if you need to track them). To do this, modify the Scan state to include a list of collected messages:

.Scan(new { CurrentStage = "Idle", CollectedMessages = new List<DeviceMessage>(), FlowReset = false }, (state, msg) =>
{
    if (msg.TypeIdentifier == "A")
    {
        var newList = new List<DeviceMessage> { msg };
        return new { CurrentStage = "WaitingForB", CollectedMessages = newList, FlowReset = true };
    }

    // Progress the flow and add the message to the collected list
    return state.CurrentStage switch
    {
        "WaitingForB" when msg.TypeIdentifier == "B" => 
            new { CurrentStage = "WaitingForC", CollectedMessages = state.CollectedMessages.Append(msg).ToList(), FlowReset = false },
        // ... repeat for C and D
        _ => state
    };
})
.Where(state => state.CurrentStage == "FlowCompleted")
.Select(state => state.CollectedMessages)
.Subscribe(flowMessages => RunPostAnalysis(flowMessages));

Final Verdict #

Rx.NET is an ideal tool for this problem—it excels at handling asynchronous, sequential event streams with state management. The above code directly addresses all your requirements, and it’s flexible enough to adapt to any additional edge cases you might encounter.

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