// {app_root}/config/plugin.js
exports.kafka = {
  enable: true,
  package: 'egg-nodekafka',
};

// {app_root}/config/config.default.js
exports.kafka = {
  client: {
    host: '127.0.0.1:9092, 127.0.0.1:9093', // 多个ip可用逗号分隔
  },
  agent: true, // 支持在agent中使用kafka对象，如agent.kafka(名称视具体配置)
  app: true, // 同上
};

'use strict';

const Controller = require('egg').Controller;

class HomeController extends Controller {

  async index() {
    const { app, logger, ctx } = this;
    const producer = app.kafka.producer(); // 此句为实例化一个producer对象，若在定时任务中注意实例化多次的异常情况
    let result;
    try {
      result = await producer.sendAsync([{
        topic: 'topic1',
        messages: 'test' + new Date().toISOString(),
        partition: 0,
      }]); 
    } catch (error) {
      logger.error(error);
      ctx.status = 201;
      return;
    }
    logger.info(result, '-------------------------');
    ctx.status = 200;
    ctx.body = { result: 'success', value: JSON.stringify(result) };
  }
}

module.exports = HomeController;

{
   topic: 'topicName',
   messages: ['message body'], // multi messages should be a array, single message can be just a string or a KeyedMessage instance
   key: 'theKey', // string or buffer, only needed when using keyed partitioner
   partition: 0, // default 0
   attributes: 2, // default: 0
   timestamp: Date.now() // <-- defaults to Date.now() (only available with kafka v0.10+)
}

// {app_root}/config/config.{$env}.js
exports.kafka = {
  client: {
    host: '127.0.0.1:9092, 127.0.0.1:9093', // 多个ip可用逗号分隔
    connectTimeout: 10000,
    requestTimeout: 30000,
    sslOptions: {},
    // 更多参考： https://github.com/SOHU-Co/kafka-node#options
  },
  agent: true, // 支持在agent中使用kafka对象，如agent.kafka(名称视具体配置)
  app: true, // 同上
};