从Kafka消费消息存入List时出现内存泄漏问题求助

问题详情 #

搭建了一个简易Web API，托管了Hosted Service，该服务包含无限循环，每15秒执行一次以下操作：从Kafka Topic拉取新记录并存入List，之后等待15秒重复。

每次循环执行后内存持续增长且从未释放，通过以下测试定位问题：

• 注释list.Add(record);后，内存泄漏消失
• 使用模拟字符串存入List无泄漏，但存入consumer.Consume().Message.Value获取的字符串时，泄漏出现

尝试过的无效处理方式：

• 每次消费循环前调用list.Clear()
• 在using语句后调用GC.Collect()
• 消费循环前将list设为null
  内存快照显示List已被清空，但内存并未释放。

复现步骤 #

创建Hosted Service，在循环中重复执行Kafka消息消费并将结果存入List的操作。

相关代码 #

Hosted Service代码 #

public class HostedMailsFromKafkaSender : BackgroundService
{
    private readonly ILogger _logger;
    private readonly IKafkaSender _kafkaSender;
    public HostedMailsFromKafkaSender(IKafkaSender sender, ILogger logger)
    {
        _kafkaSender = sender;
        _logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken cancellationToken)
    {
        await Task.Delay(5000, cancellationToken);

        while (!cancellationToken.IsCancellationRequested)
        {
            try
            {
                _kafkaSender?.SendMails();
            }
            catch (Exception e)
            {
                _logger.LogCritical(e, "Send error");
            }
            await Task.Delay(TimeSpan.FromSeconds(15), cancellationToken);
        }
    }
}

Kafka消费逻辑 #

public List ReadFromTopic(string topic)
{
    List mailListTest = new();
    var consumerConfig = new ConsumerConfig();
    config.GetSection("Kafka:ConsumerSettings").Bind(consumerConfig);
    using (var _consumer = new ConsumerBuilder<string, string>(consumerConfig).Build())
    {
        try
        {
            _consumer.Subscribe(topic);

            while (true)
            {
                var consumeResult = _consumer.Consume(10000);
                if (consumeResult.IsPartitionEOF)
                    break;

                // 尝试过去掉转换，问题依旧
                var encodingBytesRecord = Encoding.UTF8.GetBytes(consumeResult.Message.Value.ToString());
                var record = Encoding.UTF8.GetString(encodingBytesRecord);
                try
                {
                    mailListTest.Add(record); 
                }
                catch (Exception ex)
                {
                    
                }
            }
            return null;
        }
        catch (Exception ex)
        {
            return null;
        }
    }
}

DI注册代码 #

public void ConfigureServices(IServiceCollection services)
{
    services.AddTransient<IKafkaSender, KafkaSender>();
    services.AddHostedService<HostedMailsFromKafkaSender>();
    services.AddControllers();
}

环境信息 #

• Confluent.Kafka NuGet版本：1.9.2
• 消费者配置：

{
    "AutoOffsetReset": "Earliest",
    "EnableAutoCommit": "false",
    "EnablePartitionEof": "true",
    "Topic": "",
    "BootstrapServers": "",
    "SslKeystorePassword": "",
    "SslCaLocation": "",
    "SecurityProtocol": "Ssl",
    "MaxPollIntervalMs": 600000
}

• 测试操作系统：Windows 10、CentOS 8

分析与更新 #

通过dotMemory分析，发现内存增长出现在GetString方法中。已在Confluent.Kafka仓库提交相关Issue，但仍怀疑问题出在自身应用逻辑中。

更新测试：将单条Kafka记录的文本存入变量，再重复27000次存入List，未出现内存泄漏。推测问题可能出在Kafka消息类型转换为.NET string的过程中。

内容的提问来源于stack exchange，提问作者Wafeelka