消息的消费组配置、堆积总结

一、Azure EventHub
1.1、消费客户端负载均衡

引用包：Azure.Messaging.EventHubs；EventHubConsumerClient简单消费；不能持久化保存所有权\检查点，做负载均衡；需要自定义实现 EventProcessor<TPartition>，如通过Redis实现所有权获取\设置、检查点获取\设置。

参考文档：

https://devblogs.microsoft.com/azure-sdk/eventhubs-clients/

https://devblogs.microsoft.com/azure-sdk/custom-event-processor/

1.2、提供吞吐量，避免堆积

eventBatchMaximumCount ---消息批处理最大数量

EventProcessorOptions：

  PrefetchCount ---Event处理器 缓冲区预取数量

  ConnectionOptions.ReceiveBufferSizeInBytes---EventHub 接收缓冲区字节

增大这3个参数的值，可明显提高吞吐量。

protected override async Task OnProcessingEventBatchAsync(IEnumerable<EventData> events, TPartition partition, CancellationToken cancellationToken)
{
    if (events.Any())
    {
        //一次多个消息可并发处理
        
        var parallelQuery = Partitioner
        //负载平衡方式执行
        .Create(array, true)
        .GetPartitions(partitionCount)
        .AsParallel().Select(async (selector) =>
        {
            using (selector)
            {
                while (selector.MoveNext())
                {
                    byte[] eventBodyBytes = selector.Current.EventBody.ToArray();

                    string body = Encoding.UTF8.GetString(eventBodyBytes);
                    //反序列化，业务逻辑处理
                }
            }
        });
        await Task.WhenAll(parallelQuery.ToArray());

        //消费成功设置检查点，保存redis
    }
}

二、kafka
2.1、提供吞吐量，避免堆积

引用包：Confluent.Kafka，topic分片和消费组数量很多，如：12个，解决消费堆积办法：

使用自动提交策略，不要手动Commit(consumeResult)，它是阻塞方法。
ConsumerConfig：

  EnableAutoCommit(enable.auto.commit) 启动自动提交，默认true

  AutoCommitIntervalMs(auto.commit.interval.ms) 自动提交时间间隔，默认5秒

其他可配置参数(ConsumerConfig 有相对应的属性)的默认值，对吞吐量影响不大，如：

  fetch.max.bytes
  max.partition.fetch.bytes
  queued.max.messages.kbytes         
  queued.min.messages
  fetch.wait.max.ms
  receive.message.max.bytes
  socket.receive.buffer.bytes

consumerBuilder.Consume() 一次仅返回单条，先收集到ActionBlock<string> 再异步并发消费；ActionBlock的参数 ExecutionDataflowBlockOptions.MaxDegreeOfParallelism 设置并发度。

问题：应用发布时可能重复消费、消费ActionBlock异常时可能丢失消费。

适用场景：业务流程对 及时性 优先于 准确性\完整性，生产者有定时重复数据、消费保存数据使用是有降级兜底，采用自动提交消费offset的方案，提高吞吐量。