如何在Node.js中通过Kafka将三个微服务连接到API网关

整体架构逻辑 #

采用API网关作为Kafka生产者接收外部HTTP请求，按业务类型将消息转发到对应Kafka主题；各微服务作为Kafka消费者监听专属主题处理业务，完成后将结果发送到统一响应主题，网关通过唯一标识匹配请求与响应，最终返回结果给客户端。这种架构实现了网关与微服务的解耦，同时支持异步业务处理。

具体实现步骤 #

1. Kafka主题规划 #

按微服务职责拆分请求主题，同时设置统一响应主题：

• ms-authentication-requests：处理登录、鉴权等认证类请求
• ms-application-requests：处理申请提交、状态查询等业务类请求
• ms-notification-requests：处理邮件、短信发送等通知类请求
• service-responses：所有微服务返回结果的统一主题，网关通过correlationId匹配请求与响应

2. API网关端（Node.js）实现 #

首先安装Kafka客户端依赖：

npm install kafkajs express

核心逻辑代码：

const { Kafka } = require('kafkajs');
const express = require('express');
const app = express();
app.use(express.json());

// 初始化Kafka客户端
const kafka = new Kafka({
  clientId: 'api-gateway',
  brokers: ['localhost:9092'] // 替换为你的Kafka集群地址
});

const producer = kafka.producer();
const consumer = kafka.consumer({ groupId: 'gateway-response-group' });

// 存储请求等待响应的Promise映射，用correlationId做键
const responseWaitMap = new Map();

// 启动Kafka生产者与消费者
async function initKafka() {
  await producer.connect();
  await consumer.connect();
  await consumer.subscribe({ topic: 'service-responses', fromBeginning: false });
  
  // 监听响应主题，匹配请求并返回结果
  consumer.run({
    eachMessage: async ({ message }) => {
      const response = JSON.parse(message.value.toString());
      const { correlationId, data, error } = response;
      
      if (responseWaitMap.has(correlationId)) {
        const resolve = responseWaitMap.get(correlationId);
        resolve({ data, error });
        responseWaitMap.delete(correlationId);
      }
    },
  });
}

initKafka();

// 认证登录接口示例
app.post('/auth/login', async (req, res) => {
  // 生成唯一请求标识
  const correlationId = `${Date.now()}-${Math.random().toString(36).slice(2, 10)}`;
  const requestMsg = {
    correlationId,
    action: 'login',
    payload: req.body
  };
  
  // 发送请求到认证服务主题
  await producer.send({
    topic: 'ms-authentication-requests',
    messages: [{ value: JSON.stringify(requestMsg) }]
  });
  
  // 等待响应，设置超时
  const { data, error } = await new Promise(resolve => {
    responseWaitMap.set(correlationId, resolve);
    setTimeout(() => resolve({ error: '请求超时，请稍后重试' }), 10000);
  });
  
  error ? res.status(500).json({ error }) : res.json(data);
});

// 业务申请提交接口示例
app.post('/application/submit', async (req, res) => {
  const correlationId = `${Date.now()}-${Math.random().toString(36).slice(2, 10)}`;
  const requestMsg = {
    correlationId,
    action: 'submit-application',
    payload: req.body
  };
  
  await producer.send({
    topic: 'ms-application-requests',
    messages: [{ value: JSON.stringify(requestMsg) }]
  });
  
  const { data, error } = await new Promise(resolve => {
    responseWaitMap.set(correlationId, resolve);
    setTimeout(() => resolve({ error: '请求超时，请稍后重试' }), 15000);
  });
  
  error ? res.status(500).json({ error }) : res.json(data);
});

// 邮件发送接口示例
app.post('/notification/send-email', async (req, res) => {
  const correlationId = `${Date.now()}-${Math.random().toString(36).slice(2, 10)}`;
  const requestMsg = {
    correlationId,
    action: 'send-email',
    payload: req.body
  };
  
  await producer.send({
    topic: 'ms-notification-requests',
    messages: [{ value: JSON.stringify(requestMsg) }]
  });
  
  const { data, error } = await new Promise(resolve => {
    responseWaitMap.set(correlationId, resolve);
    setTimeout(() => resolve({ error: '请求超时，请稍后重试' }), 10000);
  });
  
  error ? res.status(500).json({ error }) : res.json(data);
});

app.listen(3000, () => console.log('API网关运行在端口3000'));

3. 微服务端（以ms-authentication为例）实现 #

同样安装kafkajs依赖，核心逻辑代码：

const { Kafka } = require('kafkajs');

const kafka = new Kafka({
  clientId: 'ms-authentication',
  brokers: ['localhost:9092']
});

const consumer = kafka.consumer({ groupId: 'ms-authentication-group' });
const producer = kafka.producer();

// 实际认证业务逻辑
async function handleLogin(payload) {
  const { username, password } = payload;
  // 替换为你的真实认证逻辑（比如查询数据库、生成JWT）
  if (username === 'admin' && password === 'admin123') {
    return { token: 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...' };
  }
  throw new Error('用户名或密码错误');
}

async function initService() {
  await producer.connect();
  await consumer.connect();
  await consumer.subscribe({ topic: 'ms-authentication-requests', fromBeginning: false });
  
  // 处理请求并返回响应
  consumer.run({
    eachMessage: async ({ message }) => {
      const request = JSON.parse(message.value.toString());
      const { correlationId, action, payload } = request;
      let responseData = null;
      let responseError = null;
      
      try {
        switch(action) {
          case 'login':
            responseData = await handleLogin(payload);
            break;
          case 'validate-token':
            // 扩展：添加token校验逻辑
            responseData = { valid: true };
            break;
          default:
            throw new Error(`不支持的操作类型：${action}`);
        }
      } catch (err) {
        responseError = err.message;
      }
      
      // 发送响应到统一响应主题
      await producer.send({
        topic: 'service-responses',
        messages: [{
          value: JSON.stringify({ correlationId, data: responseData, error: responseError })
        }]
      });
    },
  });
}

initService();

ms-application和ms-notification的实现逻辑完全一致，只需修改监听的主题、业务处理函数及支持的action即可。

4. 关键注意事项 #

• 唯一标识（correlationId）：必须保证每个请求的correlationId全局唯一，这是网关匹配请求与响应的核心依据。
• 超时控制：网关必须为每个请求设置超时时间，避免客户端无限等待。
• Kafka高可用：生产环境务必使用多broker集群，避免单点故障。
• 错误处理：微服务处理失败时，可将消息转发到死信队列（DLQ），便于后续排查与重试。
• 消息序列化：推荐使用JSON或Avro进行消息序列化，确保跨服务的兼容性与可读性。

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