如何向kafka发送连续的HTTP流数据？

Question

我们有一个通过 HTTP 发送时间序列数据的系统（示例系统 A）。

我现在想将 Apache Kafka 引入系统。

我的想法是让 Kafka 服务器运行，系统 A 在收到时间序列消息后立即将此消息发布到 Apache Kafka 服务器。

在这里，我有个疑问，如何发布通过HTTP协议发送的时间序列数据？

这是一个例子。

我有以下网址来收集数据：

http://ipaddress:portno/HttpPush/Track?event

接收以下流数据

2017-08-17 00:00:21 信息：?data=$Sensor,x8,86832364748057,20170816235919,13,1,1,5|2,0,7,20170816235854,0,0,0,0,128 ,0,4095|3,0,10,20170816235854,0.00,0.00,0.00,0.00,0.00,0.00,0.00,0.00,0.00,0.00|2,0,7,20170816235859,0,0,0,0,128,0 ,4095

注意：我正在接收 20kps 数据

任何人都可以帮我提供此示例的示例 kafka 生产者 API。

Answer 1

Kafka REST Proxy 在这里可能有用。源系统可以通过 HTTP 将其数据直接流式传输到 Kafka。

Answer 2

您可以将 URL 中的数据转换为 String，并将该数据复制到将数据发布到 kafka 的方法中。 1、用生产者属性初始化生产者（例如，参考下面的代码sn-p）

producer = new KafkaProducer<String, String>(KafkaConfiguration.producerProperties);
            logger.info("JetKafkaProducer{" + this.hashCode() + "} initalised Successfully ,for the topic {" + topic
                    + "}");
            logger.debug("JetKafkaProducer with hash code: {" + this.hashCode() + "}, for the topic{" + topic
                    + "} has producer properties: " + KafkaConfiguration.producerProperties);

现在发布消息

public void publishMesssage(String data) throws Exception {
    if (getTopic() != null && data != null) {
        ProducerRecord<String, String> record = new ProducerRecord<String, String>(this.getTopic(), data);
        producer.send(record); // This publishes message on given topic
        logger.info("Published record to Kafka " + data.toString());
    } else {
        logger.error("Failed to publish record to " + getTopic() + " Data: " + data.toString());
    }
    return;
}