应用场景可能不太一样解决方式也大有不同，就按照我之前的场景的统一回复下吧：

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挂掉时生产者内存溢出的三种思路：

* 1.获取当前ExecutorService线程池活动线程数，当活动线程数等于创建线程池线程数量时，表所有线程均处于阻塞状态，

* 此时return释放当前线程,不执行发送kafka; 特点:高并发时会丢失部分日志

*

* 2.当活动线程数等于创建线程池线程数量时，执行TimeUnit.MILLISECONDS.sleep(1000)

* 表示让出当前线程资源1秒，然后重新竞争发送; 特点：适用于kafka集群，或能及时恢复kafka服务的环境

*

* 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> 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);

}

}

});

}

}