2

既然我在Observable这里感冒了，我订阅了几次“分组”，为什么我不需要在这里发布？当我运行它时，我本来希望它会带来不需要的结果，但令我惊讶的是，它可以在有和没有 Publish 的情况下工作。这是为什么？

var subject = new List<string>
    {                            
    "test",                        
    "test",                 
    "hallo",
    "test",
    "hallo"
    }.ToObservable();
subject
    .GroupBy(x => x)
    .SelectMany(grouped => grouped.Scan(0, (count, _) => ++count)
         .Zip(grouped, (count, chars) => new { Chars = chars, Count = count }))
    .Subscribe(result => Console.WriteLine("You typed {0} {1} times", 
         result.Chars, result.Count));

// I Would have expect that I need to use Publish like that
//subject
//   .GroupBy(x => x)
//   .SelectMany(grouped => grouped.Publish(sharedGroup => 
//       sharedGroup.Scan(0, (count, _) => ++count)
//       .Zip(sharedGroup, (count, chars) => 
//           new { Chars = chars, Count = count })))
//   .Subscribe(result => Console.WriteLine("You typed {0} {1} times", 
//       result.Chars, result.Count));

Console.ReadLine();

编辑

正如 Paul 所注意到的，因为我们订阅了两次底层的冷可观察对象，我们应该重复这个序列两次。但是，我没有运气使这种效果可见。我试图插入调试行，但例如这只会打印一次“执行”。

var subject = new List<Func<string>>
{                            
() =>
    {
        Console.WriteLine("performing");
        return "test";
    },                        
() => "test",                 
() => "hallo",
() => "test",
() => "hallo"
}.ToObservable();


subject
    .Select(x => x())
    .GroupBy(x => x)
    .SelectMany(grouped => grouped.Scan(0, (count, _) => ++count)
            .Zip(grouped, (count, chars) => new { Chars = chars, Count = count }))
    .Subscribe(result => Console.WriteLine("You typed {0} {1} times",
            result.Chars, result.Count));

我想知道我们是否可以使我们正在处理一个冷的 observable 并且没有使用Publish(). 在另一个步骤中，我想看看Publish()（见上文）如何使效果消失。

编辑 2

正如 Paul 所建议的，我创建了一个IObservable<string>用于调试目的的自定义。但是，如果您在它的Subscribe()方法中设置断点，您会注意到它只会被命中一次。

class Program
{
    static void Main(string[] args)
    {
        var subject = new MyObservable();

        subject
            .GroupBy(x => x)
            .SelectMany(grouped => grouped.Scan(0, (count, _) => ++count)
                 .Zip(grouped, (count, chars) => new { Chars = chars, Count = count }))
            .Subscribe(result => Console.WriteLine("You typed {0} {1} times",
                 result.Chars, result.Count));

       Console.ReadLine();
   }
}

class MyObservable : IObservable<string>
{
    public IDisposable Subscribe(IObserver<string> observer)
    {
        observer.OnNext("test");
        observer.OnNext("test");
        observer.OnNext("hallo");
        observer.OnNext("test");
        observer.OnNext("hallo");
        return Disposable.Empty;
    }
}

所以对我来说，这个问题仍然悬而未决。Publish为什么我这里不需要这么冷Observable？

4

2 回答 2

6

您只使用一次基于列表的源，因此您不会在那里看到重复的订阅效果。回答您的问题的关键是以下观察：

从 GroupBy 中流出的 IGroupedObservable<K, T> 对象本身就是一个伪装的主体。

在内部，GroupBy 保留一个 Dictionary<K, ISubject<T>>。每当有消息进入时，它都会使用相应的键发送到主题中。您订阅了两次分组对象，这是安全的，因为主体将生产者与消费者分离。

于 2011-11-15T20:52:50.097 回答
1

在 Zip 中重用“分组”意味着您有效地对每个分组进行了两次 - 但是，由于您的来源是冷的，它仍然有效。那有意义吗？

于 2011-11-13T00:37:32.933 回答