Rx.NET的使用

var observable = Enumerable.Range(1, 100).ToObservable(NewTheadScheduler.Default);//申明可观察序列
Subject<int> subject = new Subject<int>();//申明Subject
subject.Subscribe((temperature) => Console.WriteLine($"当前温度：{temperature}"));//订阅subject
subject.Subscribe((temperature) => Console.WriteLine($"嘟嘟嘟，当前水温：{temperature}"));//订阅subject
observable.Subscribe(subject);//订阅observable

            Observable.Create<int>(observer =>
            {
                for (int i = 0; i < 5; i++)
                {
                    observer.OnNext(i);
                }
                observer.OnCompleted();
                return Disposable.Empty;
            }).Subscribe(o => Console.WriteLine("Here is the data " + o));

        public static void TestCreate()
        {
            IObservable<int> ob =
            Observable.Create<int>(o =>
            {
                var cancel = new CancellationDisposable(); // internally creates a new CancellationTokenSource
                NewThreadScheduler.Default.Schedule
                (
                    () =>
                    {
                    int i = 0;
                    for (; ; )
                    {
                        Thread.Sleep(200);  // here we do the long lasting background operation
                            if (!cancel.Token.IsCancellationRequested)    // check cancel token periodically
                                o.OnNext(i++);
                        else
                        {
                            Console.WriteLine("Aborting because cancel event was signaled!");
                            o.OnCompleted(); // will not make it to the subscriber
                                return;
                        }
                    }
                    }
                );

                return cancel;
            }
            );
            IDisposable subscription = ob.Subscribe(i => Console.WriteLine(i));
            Console.WriteLine("Press any key to cancel");
            Console.ReadKey();
            subscription.Dispose();
            Console.WriteLine("Press any key to quit");
            Console.ReadKey();  // give background thread chance to write the cancel acknowledge message
        }

        private static void TestRange()
        {
            IObservable<int> observable = Observable.Range(0, 10).Select(i => i * 2);
            observable.Subscribe(o => Console.WriteLine("Data is " + o));
        }
        private static void TestGenerate()
        {
            var ob=Observable.Generate(0,
                i => i < 10,
                i => i + 1,
                i=>i*2);
            ob.Subscribe(o => Console.WriteLine("Data is " + o));
        }

        public static void StartBackgroundWork()
        {
            Console.WriteLine("Shows use of Start to start on a background thread:");
            var o = Observable.Start(() =>
            {
                //This starts on a background thread.
                Console.WriteLine("From background thread. Does not block main thread.");
                PrintThreadID();
                Console.WriteLine("Calculating...");
                Thread.Sleep(3000);
                Console.WriteLine("Background work completed.");
            }).Finally(() =>
            {
                Console.WriteLine("Main thread completed.");
                PrintThreadID();
            });
            Console.WriteLine("
	 In Main Thread...
");
            PrintThreadID();
            o.Wait();   // Wait for completion of background operation.
        }