Rx .Net简介

[NullableContextAttribute(1)]
public interface IObservable<[NullableAttribute(2)] out T>
{
    IDisposable Subscribe(IObserver<T> observer);
}

[NullableContextAttribute(1)]
public interface IObserver<[NullableAttribute(2)] in T>
{
　　
    void OnCompleted();

    void OnError(Exception error);

    void OnNext(T value);
}

static async Task Main(string[] args)
{

    // 定义事件流
    var sequence = GetGenerateObservable();
    // 对事件流进行筛选
    sequence = sequence.Where(a => a < 23);

    // 注册观察者
    ReplaySubject<int> subject = new ReplaySubject<int>();//申明Subject
    subject.Subscribe((temperature) => Console.WriteLine($"当前温度：{temperature}"));//订阅subject
    subject.Subscribe((temperature) => Console.WriteLine($"嘟嘟嘟，当前水温：{temperature}"));//订阅subject
    
    // 触发
    sequence.Subscribe(subject);
    Console.ReadKey();
}

private static IObservable<int> GetSimpleObservable()
{
    return Observable.Return(42);
}

private static IObservable<int> GetThrowObservable()
{
    return Observable.Throw<int>(new ArgumentException("Error in observable"));
}

private static IObservable<int> GetEmptyObservable()
{
    return Observable.Empty<int>();
}

private static IObservable<int> GetTaskObservable()
{
    return GetTask().ToObservable();
}

private static async Task<int> GetTask()
{
    return 42;
}

private static IObservable<int> GetRangeObservable()
{
    return Observable.Range(2, 10);
}

private static IObservable<long> GetIntervalObservable()
{
    return Observable.Interval(TimeSpan.FromMilliseconds(200));
}

private static IObservable<int> GetCreateObservable()
{
    return Observable.Create<int>(observer =>
    {
        observer.OnNext(1);
        observer.OnNext(2);
        observer.OnNext(3);
        observer.OnNext(4);
        observer.OnCompleted();
        return Disposable.Empty;
    });
}

private static IObservable<int> GetGenerateObservable()
{
    // 类似for循环
    return Observable.Generate(
        1,              // 初始值
        x => x < 10,    // 循环停止条件
        x => x + 1,     // 循环一次+1
        x => x + 20     // 循环中执行的操作
    );
}