如何正确中断Kafka Producer多线程？解决关闭后发送报错问题

错误原因 #

你遇到的IllegalStateException是因为线程中断后中断状态被清除导致循环继续执行，而此时你已经在catch块中关闭了Producer，后续的producer.send()操作自然会报错。

具体逻辑问题：

• 外部调用thread.interrupt()时，Thread.sleep(10)抛出InterruptedException并进入catch块；
• Java抛出InterruptedException时会自动清除线程的中断标志，导致while (!thread.isInterrupted())条件仍为true，循环继续执行；
• 此时Producer已被关闭，下一次执行producer.send()就触发了"Cannot perform operation after producer has been closed"异常。

修复方案 #

需要恢复线程的中断状态确保循环正常退出，同时将Producer的关闭逻辑统一放在finally块中，避免重复关闭或资源泄漏。

修正后的代码：

@Override
public void run() {
    logger.info("Producer Thread started for: " + this.topic);
    ProducerRecord<byte[], byte[]> record;
    try {
        while (!Thread.currentThread().isInterrupted()) {
            Thread.sleep(10);
            record = new ProducerRecord<>(this.topic, new Payload());
            producer.send(record);
        }
    } catch (InterruptedException e) {
        logger.info("KafkaTupleProducer with topic " + this.topic +" interrupted");
        // 恢复线程中断状态，确保循环退出
        Thread.currentThread().interrupt();
    } finally {
        // 统一处理Producer的flush和关闭，保证资源释放
        try {
            producer.flush();
        } finally {
            producer.close();
        }
    }
}

额外优化建议 #

• 共用Producer实例：Kafka Producer本身是线程安全的，多个发送线程完全可以共用同一个Producer实例，这样能减少TCP连接数和资源占用，大幅提升发送性能，无需为每个线程单独创建Producer；
• 避免重复操作：finally块能保证无论循环正常退出还是中断退出，Producer只会被关闭一次，避免重复调用close()引发的潜在问题。

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