我已经用express.js创建了一个restapi并想连接kafka。我的 KafkaService 等级如下：

var express = require('express');
var kafka = require('kafka-node');

class KafkaServices{
    constructor(){
        this.Producer = kafka.Producer,
        this.client = new kafka.KafkaClient();
        this.producer = new kafka.Producer(this.client);
        this.payloads = [];

        if(!!KafkaServices.instance){
            return KafkaServices.instance;
        }

        KafkaServices.instance = this;
        return this;
    }
    connect(){
        // Create a connection
        this.producer.on('ready', function() {
            console.log('Kafka connection established');
        })
    }
    error(){
        // Check for Errors
        this.producer.on('error', function(err) {
            console.log('Kafka has the following error');
            console.log(err);
        })
    }
    setPayload(payload){
        // Set paylaod
        this.payloads = payload
    }
    publish(){
        //Publish
        this.producer.send(this.payload, function (err, data) {
        console.log('Kafka has published the following payload:');
        console.log(data);
        console.log(err);
        });
    }
}

module.exports = KafkaServices;

在 server.js 我正在访问 KafkaServices 要连接到kafka并在连接不成功时检查错误，请执行以下操作：

var KafkaServices = require('./services/kafka-services');
kafka = require('kafka-node');
function onListening() {
    var addr = server.address();
    var bind = typeof addr === 'string'
      ? 'pipe ' + addr
      : 'port ' + addr.port;
    console.log('Listening on ' + bind);
    logInfo({message: `Server is Listening on http://localhost:${addr.port}.`, scope: 'Server'});
    kafkaTool = new KafkaServices();
    // Connect to kafka 
    kafkaTool.connect();
    kafkaTool.error();
}

在 settings-router.js 我有我的post方法，在该post方法中，我想将有效负载发布到kafka，如下所示：

const express = require('express');
const router = express.Router();
var kafka = require('kafka-node');
const KafkaServices = require('../services/kafka-services');
const { settings } = require('../app');

router.post('/', (req, res, next) => {
    console.log("Posting...");
    try{
        kafkaTool = new KafkaServices();
        kafkaTool.setPayload(req.body);
        kafkaTool.publish();
        console.log('Payload has been published to Kafka');
    }
    catch(err){
        console.log('Error Caught');
        console.log(req.body);
        console.log(err);
    }
    return res.status(200).json({
        message: 'Setting details',
        payload: req.body
    });
});

module.exports = router;

当前，当我尝试发布时，出现以下错误：

TypeError: Cannot read property 'forEach' of undefined
at Producer.BaseProducer.buildPayloads (C:\Users\ENV\Projects\settingtool\node_modules\kafka-node\lib\baseProducer.js:125:12)
at Producer.BaseProducer.send (C:\Users\ENV\Projects\settingtool\node_modules\kafka-node\lib\baseProducer.js:120:34)
at KafkaServices.publish (C:\Users\ENV\Projects\settingtool\src\services\kafka-services.js:45:23)
at router.post (C:\Users\ENV\Projects\settingtool\src\routes\settings-router.js:12:19)
at Layer.handle [as handle_request] (C:\Users\ENV\Projects\settingtool\node_modules\express\lib\router\layer.js:95:5)
at next (C:\Users\ENV\Projects\settingtool\node_modules\express\lib\router\route.js:137:13)
at Route.dispatch (C:\Users\ENV\Projects\settingtool\node_modules\express\lib\router\route.js:112:3)
at Layer.handle [as handle_request] (C:\Users\ENV\Projects\settingtool\node_modules\express\lib\router\layer.js:95:5)
at C:\Users\ENV\Projects\settingtool\node_modules\express\lib\router\index.js:281:22
at Function.process_params (C:\Users\ENV\Projects\settingtool\node_modules\express\lib\router\index.js:335:12)

我不完全确定如何克服这个错误，特别是因为这个错误是指 node_modules 是从哪来的 npm install node-kafka . 如有任何建议，将不胜感激。