Rest APIKafka流

Question

{"topic":"twitterFeeds","value":"RT @flyingtsunami: @SethAbramson **READ THIS UNTIL IT SINKS IN** #POTUS  #Comey's firing (he was); T…","offset":0,"partition":0,"highWaterOffset":1906,"key":null}
{"topic":"twitterFeeds","value":"RT @RCorbettMEP: Why does the @BBC news only …","offset":1,"partition":0,"highWaterOffset":1906,"key":null}

[
{"topic":"twitterFeeds","value":"RT @flyingtsunami: @SethAbramson **READ THIS UNTIL IT SINKS IN** #POTUS  #Comey's firing (he was); T…","offset":0,"partition":0,"highWaterOffset":1906,"key":null}
,
{"topic":"twitterFeeds","value":"RT @RCorbettMEP: Why does the @BBC news only …","offset":1,"partition":0,"highWaterOffset":1906,"key":null}
]

const express = require('express');
const router = express.Router();
    var output1 = '';
var http = require('http');  
var kafka = require('kafka-node'),
    Consumer = kafka.Consumer,
    client = new kafka.Client(),
    consumer = new Consumer(
        client,
        [
        { topic: 'twitterFeeds', partition: 0 }
        ],
        {
            autoCommit: false
        }
    );
    var output2 = consumer.on('message', function (message) {
     obj = JSON.stringify(message);
     output1 = output1 + obj;
    });

router.get('/',(req,res)=>{
        res.send(output1);
    });
module.exports = router;

Answer 1

我目前正在做一个类似的项目（如果你有兴趣，请帮忙）-超燃冲压发动机-Kafka

我已经在scramjet Datastream.tojsonArray中解决了上述问题--它只是将数据作为JSON数组流式传输。在您的情况下，它非常简单：

new ConsumerStream(
    client,
    [ { topic: 'twitterFeeds', partition: 0 } ],
    { autoCommit: false }
)
    .pipe(new scramjet.DataStream)
    // you may want to do some transforms here
    .toJSONArray()
    .pipe(process.stdout)
    // or pipe it to any other stream, accumulate and so on.

如果你喜欢这个项目，你可以帮助-请通过GitHub联系。