使用计时器和Rx生成一个整数列表

Question

这对其他人来说似乎是一个基本问题，但由于我刚刚开始使用Rx，我想知道是否有人可以为我指出正确的方向。我正在尝试订阅一个由计时器更新的流。我有一些测试代码下面应该希望产生大量的整数。但是，尽管填充了列表，但从未调用Write处理程序，因此没有输出。代码只是一个测试代码，所以我没有处理任何东西。最后，我想通过轮询和处理重复项来获取目录的内容。

using System;
using System.Collections.Generic;
using System.Reactive.Linq;
using System.Timers;
using Timer = System.Timers.Timer;
namespace ConsoleApplication1
{
    public class ObservableTest
    {
        private readonly List<int> m_numbers = new List<int>();
        private readonly Random m_random = new Random();
        public ObservableTest()
        {
            var timer = new Timer(1000);
            timer.Elapsed += MTimerOnElapsed;
            timer.Enabled = true;
        }
        private void MTimerOnElapsed(object sender, ElapsedEventArgs elapsedEventArgs)
        {
            var randomNumber = m_random.Next(0, 100);
            m_numbers.Add(randomNumber);
        }
        public IObservable<int> GetSequence()
        {
            return m_numbers.ToObservable();
        }
    }
    class Program
    {
        static void Main(string[] args)
        {
            var observable = new ObservableTest();
            observable.GetSequence().Subscribe(Write);
            Console.ReadLine();
        }
        private static void Write(int obj)
        {
            Console.WriteLine(obj);
        }
    }
}

Answer 1

在上面的代码中，m_numbers.ToObservable()行创建了一个可观察的序列，它直接遍历m_numbers中的所有值，然后调用OnCompleted。因为在第一次调用时没有值，所以什么也没发生。您可以通过在调用订阅之前添加System.Threading.Thread.Sleep(3100)来查看这一点。这将打印出三个数字。

一个更好的方法是使用Rx的一个内置方法来创建序列。一种间隔一秒获得随机数流的方法是使用Observable。间隔:

static void Main(string[] args)
{
    var random = new Random();
    var source = Observable
         .Interval(TimeSpan.FromSeconds(1))
         .Select(_ => random.Next(1, 100));
    source.Subscribe(Console.WriteLine);
    Console.ReadLine();
}

select函数忽略Interval的值(这只是一个计数器)并返回一个随机值。

Answer 2

看看这个版本

internal class Program
{
    private static readonly Random _random = new Random();
    private static void Main(string[] args)
    {
        IDisposable subscription = TimeSequence();
        Console.ReadLine();
        subscription.Dispose();
    }
    private static IDisposable TimeSequence()
    {
        IObservable<long> timer = Observable.Timer(TimeSpan.Zero, TimeSpan.FromSeconds(1));
        return timer.Subscribe(GenerateRandom);
    }
    private static void GenerateRandom(long tick)
    {
        int randomNumber = _random.Next(0, 100);
        Console.WriteLine(randomNumber);
    }
}

下面是关于创建可观察序列的更多内容，也非常有用- 101个Rx示例