Elastic.Channels 0.7.0

Elastic.Channels

提供了一种专门的System.Threading.Channels.ChannelWriter实现，使得批量消费推送到该线程的数据变得容易。

批处理将在达到一定最大值或批次的生命周期超过一定年龄时发出。

这允许以相同的方式推送不同速率的数据，同时可以实施不同的实现来将批处理数据发送给接收器。

此包主要作为核心库以及抽象类提供，不提供任何有用的实现。

它包含一个NoopBufferedChannel实现，该实现在其单元测试和基准测试目的的` exposImport 实现`中不执行任何操作。

BufferedChannelBase<>

一个抽象类，要求实现者实现

protected abstract Task<TResponse> Export(IReadOnlyCollection<TEvent> buffer, CancellationToken ctx);

任何实现都允许通过以下方式将其数据推送到它

var e = new TEvent();
if (await channel.WaitToWriteAsync(e))
	written++;

ChannelOptionsBase<>

BufferedChannelBase<>的实现者必须创建自己的ChannelOptionsBase<>实现。这样，每个通道实现都会创建一个适当命名的选项类。

简单的最小实现

public class Event { }
public class Response { }

public class NoopChannelOptions 
  : ChannelOptionsBase<Event, Response> { }

public class NoopBufferedChannel 
  : BufferedChannelBase<NoopChannelOptions, Event, Response>
{

  public NoopBufferedChannel(NoopChannelOptions options) 
    : base(options) { }

  protected override Task<Response> Export(IReadOnlyCollection<NoopEvent> buffer, CancellationToken ctx)
  {
    return Task.FromResult(new Response());
  }
}

现在一旦我们实例化了一个 NoopBufferedChannel，我们就可以用它推送数据。

var e = new Event();
if (await noopChannel.WaitToWriteAsync(e))
	written++;

为了测试或调试目的，既存在 NoopBufferedChannel，也存在更专门的 DiagnosticsBufferedChannel。

DiagnosticsBufferedChannel.ToString() 揭示了大量关于状态机的见解。

缓冲区选项

每个 ChannelOptionsBase<> 实现都会接收并公开一个 BufferOptions 实例。这控制了 BufferedChannelBase<> 的缓冲行为。