使用计时器和 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.Interval：

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);
    }
}

这里有更多关于Creating an Observable sequence的信息，也非常有用 - 101 Rx Samples