如何在Rx的Observable链中复用对象实例？

在Rx里处理这种需要复用前一步实例的场景，最常用的方式是借助**SelectMany**操作符（也就是Rx中的flatMap）来传递上下文实例，同时保持链式操作的流畅性。下面是对应你Async/Await代码的Observable实现：

基础实现（对应原Async/Await逻辑） #

public IObservable<IList<string>> Delete(IList<string> ids)
{
    // 将异步的CreateContext转换为Observable序列
    return Observable.FromTask(() => _contextFactory.CreateContext())
        .SelectMany(context =>
        {
            // 复用context实例执行删除逻辑
            var entitiesToRemove = context.Set<T>().Where(item => ids.Contains(item.Id));
            context.Set<T>().RemoveRange(entitiesToRemove);
            
            // 对应原代码直接返回ids，这里返回一个立即完成的Observable
            return Observable.Return(ids);
        });
}

包含保存更改的完善实现 #

你原Async/Await代码里没写SaveChangesAsync()，但实际业务中通常需要持久化删除操作，这里补充上Rx风格的异步保存逻辑：

public IObservable<IList<string>> Delete(IList<string> ids)
{
    return Observable.FromTask(() => _contextFactory.CreateContext())
        .SelectMany(context =>
        {
            var entitiesToRemove = context.Set<T>().Where(item => ids.Contains(item.Id));
            context.Set<T>().RemoveRange(entitiesToRemove);
            
            // 将SaveChangesAsync转换为Observable，完成后映射返回ids
            return Observable.FromTask(() => context.SaveChangesAsync())
                .Select(_ => ids);
        });
}

最佳实践：自动管理Context生命周期 #

如果你的Context实现了IDisposable，推荐用Observable.Using来自动管理资源，避免手动Dispose可能带来的泄漏问题：

public IObservable<IList<string>> Delete(IList<string> ids)
{
    // Using操作符会在序列完成/出错/取消时自动释放Context
    return Observable.Using(
        resourceFactory: () => _contextFactory.CreateContext(),
        observableFactory: context =>
        {
            var entitiesToRemove = context.Set<T>().Where(item => ids.Contains(item.Id));
            context.Set<T>().RemoveRange(entitiesToRemove);
            
            return Observable.FromTask(() => context.SaveChangesAsync())
                .Select(_ => ids);
        });
}

关键逻辑说明 #

• Observable.FromTask()：这是Rx与.NET异步代码交互的标准方式，能把Task<T>类型转换为Observable序列，让异步操作融入Rx的链式流中。
• SelectMany()：核心操作符，它允许你接收前一步产生的context实例，在其内部执行后续业务逻辑，最后返回新的Observable。和Select()不同，它会“展平”嵌套的Observable，保证整个操作是连续的异步流。
• Observable.Using()：专门用于管理可释放资源的生命周期，比手动调用Dispose()更可靠，能覆盖序列终止的所有场景（完成、错误、取消订阅）。

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