如何在KafkaJS中检查Kafka生产者是否已连接？

我有一个运行在AWS Lambda上、处理HTTPS请求的Node.js服务，使用KafkaJS操作Kafka。为了避免每次请求都初始化并连接生产者，我在容器级别初始化了生产者，希望只在生产者未连接时才重新连接，但不知道如何正确判断生产者的连接状态。当前代码中尝试使用producer._isConnected始终返回undefined，producer.isConnected()也不是KafkaJS提供的方法。

当前代码片段：

const kafka = new Kafka(kafkaConfig);
const producer = kafka.producer();

const connectProducer = async () => {
  if (!producer._isConnected) {
    logging.log("Connecting a kafka producer");
    await producer.connect();
    logging.log("Kafka producer connected successfully");
  }
};

const sendMessageToKafka = async (message, topic_name) => {
  try {
    await connectProducer();
    logging.log("Sending a message to Kafka");
    await producer.send({
      topic: topic_name,
      messages: [message],
      timeout: 30000
    });
    logging.log("Finished sending message to Kafka");
  }
  catch (error) {
    throw logging.createError(error, "Failed to send message to Kafka", 503);
  }
};

KafkaJS并没有对外暴露直接判断生产者连接状态的API，也不建议依赖私有属性（比如_isConnected），因为这类属性可能会在版本迭代中变更。推荐通过以下两种方式实现需求：

方式1：自行维护连接状态变量 #

定义一个全局变量记录连接状态，结合KafkaJS的事件监听更新状态，确保状态准确：

const kafka = new Kafka(kafkaConfig);
const producer = kafka.producer();
let isProducerConnected = false;

// 监听生产者就绪事件，标记连接状态为已连接
producer.on('ready', () => {
  isProducerConnected = true;
  logging.log("Kafka producer connected successfully");
});

// 监听生产者断开事件，标记连接状态为未连接
producer.on('disconnected', () => {
  isProducerConnected = false;
  logging.log("Kafka producer disconnected");
});

const connectProducer = async () => {
  if (!isProducerConnected) {
    logging.log("Connecting a kafka producer");
    try {
      // connect方法本身是幂等的，即使重复调用也不会重复建立连接
      await producer.connect();
    } catch (error) {
      logging.log("Failed to connect Kafka producer", error);
      throw error;
    }
  }
};

// sendMessageToKafka函数保持不变
const sendMessageToKafka = async (message, topic_name) => {
  try {
    await connectProducer();
    logging.log("Sending a message to Kafka");
    await producer.send({
      topic: topic_name,
      messages: [message],
      timeout: 30000
    });
    logging.log("Finished sending message to Kafka");
  }
  catch (error) {
    throw logging.createError(error, "Failed to send message to Kafka", 503);
  }
};

方式2：直接调用connect()（简化版） #

KafkaJS的producer.connect()方法是幂等的——如果生产者已经处于连接状态，再次调用该方法不会重复建立连接，也不会抛出错误。因此可以省略状态判断，直接在发送消息前调用await producer.connect()：

const kafka = new Kafka(kafkaConfig);
const producer = kafka.producer();

const sendMessageToKafka = async (message, topic_name) => {
  try {
    // 无需判断状态，直接调用connect
    await producer.connect();
    logging.log("Sending a message to Kafka");
    await producer.send({
      topic: topic_name,
      messages: [message],
      timeout: 30000
    });
    logging.log("Finished sending message to Kafka");
  }
  catch (error) {
    throw logging.createError(error, "Failed to send message to Kafka", 503);
  }
};

注意事项 #

在AWS Lambda环境中，容器可能会被冻结或回收，导致生产者连接断开。通过事件监听维护状态的方式，能在连接断开时及时更新状态，确保下次请求触发时自动重连。

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