如何创建Node.js的Kafka API供React Native应用调用

你需要用Node.js搭建一个API代理层承接React Native的请求，再对接Kafka集群完成消息投递，整体链路为：
React Native端 → Node.js API服务 → Kafka集群
这个方案完全规避了React Native无法直接对接Kafka的问题，同时也避免了直接把Kafka集群暴露在公网的安全风险。

1. Node.js服务端环境搭建 #

优先选用Express（轻量易上手）做API框架，kafkajs做Kafka客户端，需要的依赖安装命令：

npm install express kafkajs cors

首先初始化Kafka生产者示例代码：

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

// Kafka配置
const kafka = new Kafka({
  clientId: 'meeting-notification-producer',
  brokers: ['你的Kafka集群地址1:9092', '你的Kafka集群地址2:9092'] // 替换为实际地址
})
const producer = kafka.producer()

// 服务启动时先连接Kafka生产者
const startServer = async () => {
  await producer.connect()
  app.listen(3000, () => {
    console.log('Node.js服务运行在端口3000')
  })
}
startServer()

2. 开发供React Native调用的POST接口 #

暴露专用接口接收会议ID参数，调用Kafka生产者投递消息到指定Topic：

// 投递会议ID的接口
app.post('/api/v1/send-meeting-id', async (req, res) => {
  try {
    const { meetingId, targetUserId } = req.body
    // 基础参数校验
    if (!meetingId || !targetUserId) {
      return res.status(400).json({ code: 400, msg: '缺少会议ID或目标用户ID' })
    }
    // 发送消息到Kafka的meeting_notification主题，可根据业务调整主题名
    await producer.send({
      topic: 'meeting_notification',
      messages: [
        {
          key: targetUserId, // 用目标用户ID做key，保证同用户的消息有序
          value: JSON.stringify({ meetingId, targetUserId, sendTime: Date.now() })
        }
      ]
    })
    res.status(200).json({ code: 200, msg: '会议ID投递成功' })
  } catch (err) {
    console.error('消息投递失败', err)
    res.status(500).json({ code: 500, msg: '消息投递失败，请稍后重试' })
  }
})

3. React Native端调用逻辑 #

不需要安装任何Kafka相关依赖，直接用fetch或者axios调用Node.js接口即可：

// React Native端调用示例，生成会议ID后触发
const sendMeetingId = async (meetingId, targetUserId) => {
  try {
    const response = await fetch('http://你的Node.js服务地址:3000/api/v1/send-meeting-id', {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        // 如果有身份校验，这里加Authorization头传JWT
      },
      body: JSON.stringify({
        meetingId,
        targetUserId
      })
    })
    const result = await response.json()
    if (result.code === 200) {
      console.log('会议ID发送成功')
    }
  } catch (err) {
    console.error('请求失败', err)
  }
}

• 接口增加JWT身份校验，只允许登录后的合法用户调用，避免接口被恶意滥用
• 根据业务对消息可靠性的要求调整Kafka生产者的acks配置，需要最高可靠性可设为acks: "all"，兼顾性能可设为acks: 1
• 增加消息发送失败重试机制，以及死信队列存储多次投递失败的消息方便排查
• 接口增加请求限流规则，避免短时间大量请求压垮Kafka集群
• 公网部署时Node.js服务建议加HTTPS证书，保证传输数据安全

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