如何通过Node.js的Kafka消费者将JSON数据按模块写入MongoDB不同集合

前置准备 #

先安装所需依赖：

npm install kafkajs mongoose

同时确认以下服务已就绪：

• Kafka服务正常运行，已创建接收JSON数据的主题
• MongoDB服务正常运行，可正常连接

实现步骤 #

1. 连接MongoDB，分别定义Bin_details、Device_Status对应两个集合的Schema和数据模型
2. 初始化Kafka消费者，配置broker地址、消费组ID，订阅对应数据主题
3. 拉取到消息后先解析为JSON数组，遍历每条数据分别拆分两个模块的内容
4. 兼容处理数据中部分Bin_details字段带反引号的异常格式，分别写入对应集合
5. 增加异常捕获逻辑，避免单条数据错误导致消费进程崩溃

完整代码示例 #

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

// 连接MongoDB，替换为实际的MongoDB连接地址、库名
mongoose.connect('mongodb://localhost:27017/device_data_db')
  .then(() => console.log('MongoDB连接成功'))
  .catch(err => console.error('MongoDB连接失败', err))

// 定义Bin_details对应集合的Schema
const binDetailsSchema = new mongoose.Schema({
  deviceId: String,
  date: String
})
// 第三个参数指定MongoDB中实际的集合名
const BinDetailsModel = mongoose.model('BinDetails', binDetailsSchema, 'bin_details')

// 定义Device_Status对应集合的Schema
const deviceStatusSchema = new mongoose.Schema({
  deviceId: String,
  filledStatus: Number,
  timeStamp: String,
  date: String
})
const DeviceStatusModel = mongoose.model('DeviceStatus', deviceStatusSchema, 'device_status')

// 初始化Kafka消费者，替换为实际的Kafka broker地址
const kafka = new Kafka({
  clientId: 'device-data-consumer',
  brokers: ['localhost:9092']
})
const consumer = kafka.consumer({ groupId: 'device-data-save-group' })

const runConsumer = async () => {
  // 订阅对应主题，fromBeginning按需配置是否消费历史数据
  await consumer.subscribe({ topic: 'your_device_topic', fromBeginning: false })

  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      try {
        // 解析Kafka消息内容为JSON
        const dataList = JSON.parse(message.value.toString())

        // 批量处理数据，性能要求高可替换为bulkWrite批量写入
        for (const item of dataList) {
          // 兼容字段带反引号的异常情况
          const binInfo = item['Bin_details'] || item['`Bin_details`']
          const statusInfo = item['Device_Status']
          const recordDate = item.date

          // 写入bin_details集合
          await BinDetailsModel.create({
            deviceId: binInfo.Device_ID,
            date: recordDate
          })

          // 写入device_status集合
          await DeviceStatusModel.create({
            deviceId: binInfo.Device_ID,
            filledStatus: Number(statusInfo.Filled_status),
            timeStamp: statusInfo.TimeStamp,
            date: recordDate
          })
        }

        // 按需开启手动提交偏移量
        // await consumer.commitOffsets([{ topic, partition, offset: message.offset }])
        console.log(`批次数据处理完成，偏移量: ${message.offset}`)
      } catch (err) {
        console.error('数据处理失败:', err)
        // 生产环境可将错误消息写入死信队列，后续人工补处理
      }
    }
  })
}

runConsumer().catch(console.error)

注意事项 #

• 数据量较大的场景建议用bulkWrite代替逐条create，能大幅提升写入性能
• 可根据业务查询需求，给两个集合的deviceId、date字段加联合索引，优化查询速度
• 生产环境建议配置消费重试、死信队列逻辑，避免数据丢失

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