Kafka内存溢出

红太狼 2023-06-07 12:23 2阅读 0赞

1、内存溢出的原因：

当kafka集群(或单机)服务挂了，生产者继续向kafka发送消息时，有两个超时设置会导致线程不被及时释放，另外还有一个缓冲区大小的设置也会导致异常抛出，三个参数分别如下：

*  max.block.ms：指定生产者调用send()方法或使用partitionsFor()方法获取元数据时的阻塞时间，默认值60000ms(60秒)；
 *  request.timeout.ms：指定了生产者在发送数据时等待服务器返回响应的时间，默认值30000ms(30秒)；
 *  buffer.memory：设置生产者内存缓冲区的大小，生产者用它缓冲要发送到服务器的消息，如果应用程序发送消息的速度超过发送到服务器的速度，会导致生产者空间不足。

即使用默认配置，当kafka挂了，线程调用send()方法向kafka发送消息至少会被阻塞60s，线程分分钟就会全部被阻塞，web容器在没有可用线程时收到的请求一般还会存放在队列中等待响应，线程得不到释放意味着内存同样无法被释放，所以很快内存就溢出了。

解决思路：

因此适当减少阻塞超时时长（测试设置为300ms）、增加生产者内存缓冲区，即便kafka挂了只要能即时释放线程及内存，应用服务就不至于挂掉，但阻塞时长过小有可能导致kafka网络波动时部分数据丢失，对数据有严格要求的场景并不适用。另外也可以从线程池里做限制，避免高并发场景下线程堵死的情况。

样例代码及分析：

package com.demo.kafka.util;
    
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.kafka.support.SendResult;
    import org.springframework.util.concurrent.ListenableFuture;
    import com.demo.kafka.util.KafkaProducerFactory;
    
    public class KafkaAppender{
    private static final Logger log = LoggerFactory.getLogger(KafkaAppender.class);
    // 线程数
    private static int threadPoolNum = 20;
    private static ExecutorService exec = Executors.newFixedThreadPool(threadPoolNum);
    
    /**
     * 解决当kafka挂掉时生产者内存溢出的三种思路：</p>
     * 1.获取当前ExecutorService线程池活动线程数，当活动线程数等于创建线程池线程数量时，表所有线程均处于阻塞状态，
     * 此时return释放当前线程,不执行发送kafka; 特点:高并发时会丢失部分日志</p>
     * 
     * 2.当活动线程数等于创建线程池线程数量时，执行TimeUnit.MILLISECONDS.sleep(1000)
     * 表示让出当前线程资源1秒，然后重新竞争发送; 特点：适用于kafka集群，或能及时恢复kafka服务的环境</p>
     * 
     * 3.设置最大阻塞时长max.block.ms，和最大请求时长request.timeout.ms为300ms（默认60s）
     * 表示执行发送kafka超过300ms即认为发送失败，直接结束当前线程；(测试平均发送一条日志耗时为5ms)
     * 
     */
    public void sendMsg2Kafka(final String msg) {
    
        // 查询当前线程池活动线程数
        int threadNum = ((ThreadPoolExecutor) exec).getActiveCount();
        log.info("当前线程池活动线程数:{}", threadNum);
    
        // 1(供参考).当线程池没有可用线程时接结束该线程任务(丢弃日志)
    // while (threadNum == threadPoolNum) {
    // log.info("线程池可用线程数为0,丢弃该条日志");
    // return;
    // }
    
        // 2(供参考).当线程池没有可用线程时，调用线程进入睡眠状态，并让出执行机会给其它线程1000ms
    // while (threadNum == threadPoolNum) {
    // try {
    // log.info("调用线程进入睡眠1000ms");
    // TimeUnit.MILLISECONDS.sleep(1000);
    // } catch (InterruptedException e) {
    // log.error("Exception:", e);
    // }
    // }
    
        // 3. 发送消息到kafka
        exec.execute(new Runnable() {
            @Override
            public void run() {
                long start = System.nanoTime();
                try {
                    // 日志消息发送到kafka，并获获取返回结果
                    // KafkaProducerFactory是手动封装的一个获取KafkaTemplate的工厂类
                    ListenableFuture<SendResult<String, String>> result = KafkaProducerFactory.getKafkaTemplate().sendDefault(msg);
    
                    // 解析回调函数确认是否发送成功,失败时打印失败信息及阻塞时长
                    if (result != null) {
                        Long offsetIndex = result.get().getRecordMetadata().offset();
                        if (offsetIndex != null && offsetIndex >= 0) {
                            // 发送成功
                            long end = System.nanoTime();
                            log.info("日志发送成功，offset：{}，发送耗时：{}ms", offsetIndex,TimeUnit.NANOSECONDS.toMillis(end - start));
                        } else {
                            // 发送失败
                            long end = System.nanoTime();
                            log.info("日志发送失败，阻塞时长:{}ms", TimeUnit.NANOSECONDS.toMillis(end - start));
                        }
                    }
                } catch (Exception e) {
                    // 发送异常
                    long end = System.nanoTime();
                    log.error("日志发送异常，阻塞时长:{}ms", TimeUnit.NANOSECONDS.toMillis(end - start), e);
                }
            }
        });
    }
    }