按组在单独的线程中运行处理

Question

我正在尝试在我的 Kafka 消费者中使用 Rx。

public static event EventHandler<ConsumeResult<string, string>> GenericEvent;

然后我有以下代码

var observable = Observable.FromEventPattern<ConsumeResult<string, string>>(
     x => GenericEvent += x,
     x => GenericEvent -= x)
  .Select(x => x.EventArgs);

while (!cancellationToken.IsCancellationRequested)
{
    ConsumeResult<string, string> consumeResult = _consumer.Consume(cancellationToken);
    GenericEvent(consumeResult.Topic, consumeResult);
}

然后我在某处使用它

var subscription = observable.Subscribe(message =>
{
    Console.WriteLine($"{Thread.CurrentThread.ManagedThreadId} ** {message.Topic}/{message.Partition} @{message.Offset}: '{message.Value}'");

    //_kafkaConsumer.Commit(messages);
});

是否有可能按主题名称（consumeResult.Topic）运行单独的线程？消费者收到消息后，按主题将消息重定向到对应的线程

Answer 1

试一试：

Observable
    .Interval(TimeSpan.FromSeconds(0.1))
    .Take(20)
    .GroupBy(x => x % 3)
    .SelectMany(xs => Observable.Using(() => new EventLoopScheduler(), els => xs.ObserveOn(els)))
    .Subscribe(x => Console.WriteLine($"{x} {Thread.CurrentThread.ManagedThreadId}"));

这确保在new EventLoopScheduler() 调度程序中为GroupBy 运算符创建的每个内部可观察对象创建一个线程。 SelectMany 扁平化组，但保持EventLoopScheduler 与每个组相关联。

在你的情况下，你 GroupBy consumeResult.Topic 属性。

请确保您的源 observable 在线程永远存在之前结束。在订阅上调用 Dispose() 足以结束 observable。