如何基于控制台输入创建IObservable<string>并解决订阅相关问题

• 原FromConsole实现中Console.ReadLine()是同步阻塞方法，第一个订阅触发后会直接卡在调用上，后续订阅逻辑无法正常执行
• 自定义实现默认是冷可观察对象，每个订阅都会启动独立的读行循环，但系统控制台输入是全局独占资源，多个读行调用会互相抢占，无法拿到同一份输入
• 即便加了Publish().RefCount()，第一个订阅触发的阻塞读逻辑也会卡死整个订阅流程，无法实现多播效果

核心思路是将控制台读逻辑做成单例后台任务，用Subject做中间层实现多播：

1. 只启动一次后台长活任务读取控制台输入，避免多订阅抢占输入资源
2. 用Subject<string>接收输入并广播给所有订阅者，实现同一份输入多分发
3. 读行逻辑放在独立后台线程运行，不阻塞主线程和订阅流程

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

static class Program
{
    private static readonly Subject<string> _consoleInputSubject = new();
    private static bool _readLoopStarted;
    
    static async Task Main(string[] args)
    {
        var observable = FromConsole().Publish().RefCount();
        observable.Subscribe(input => Console.WriteLine($"订阅1输出：{input}"));
        await Task.Delay(1500);
        observable.Subscribe(input => Console.WriteLine($"订阅2输出：{input}"));
        await new TaskCompletionSource().Task;
    }

    static IObservable<string> FromConsole()
    {
        // 确保读循环仅启动一次
        if (!_readLoopStarted)
        {
            StartReadLoop();
            _readLoopStarted = true;
        }
        return _consoleInputSubject.AsObservable();
    }

    private static void StartReadLoop()
    {
        // 后台长活线程运行读逻辑，不阻塞主线程
        _ = Task.Factory.StartNew(() =>
        {
            while (true)
            {
                var input = Console.ReadLine();
                if (input != null)
                {
                    _consoleInputSubject.OnNext(input);
                }
                // 可在此处增加控制台关闭判断，调用OnCompleted结束可观察流
            }
        }, TaskCreationOptions.LongRunning);
    }
}

修改后运行效果和Observable.Interval一致，同一次控制台输入会触发所有订阅的回调，无线程阻塞问题。

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