KafkaClient：

概念：能夠直接連接Kafka brokers的client
初始化：const client = new kafka.KafkaClient({kafkaHost: '10.3.100.196:9092'}); 沒有填寫kafkaHost默認(rèn)是：localhost:9092

Producer

初始化：Producer(KafkaClient, [options], [customPartitioner])
例：

var kafka = require('kafka-node'),
    Producer = kafka.Producer,
    client = new kafka.KafkaClient(),
    producer = new Producer(client);

發(fā)送消息：
send(payloads, cb)

payloads：數(shù)組，item形如json
{
   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+)
}
cb: 處理成功或失敗的回調(diào)函數(shù)

創(chuàng)建Topic
createTopics(topics, cb)

HighLevelProducer

HighLevelProducer(KafkaClient, [options], [customPartitioner])
send(payloads, cb)
createTopics(topics, async, cb)

ProducerStream

ProducerStream (options)

• highWaterMark size of write buffer (Default: 100)
• kafkaClient options see KafkaClient
• producer options for Producer see HighLevelProducer

案例：
使用Transform去更新數(shù)據(jù)

Consumer

Consumer(client, payloads, options)
on('error', function (err) {})
on('offsetOutOfRange', function (err) {})

addTopics(topics, cb, fromOffset)
removeTopics(topics, cb)
commit(cb)
setOffset(topic, partition, offset)
pause()
resume()
pauseTopics(topics)
resumeTopics(topics)
close(force, cb)

ConsumerStream

ConsumerStream(client, payloads, options)

ConsumerGroup

ConsumerGroup(options, topics)