javascript – Spring Websocket与Kafka集成

2023年4月27日 391次阅读

我正在尝试通过
Spring MVC项目中的
Spring-Websockets将消耗的Kafka数据发送到前端(
JavaScript).

为了建立服务器和客户端之间的通信,我有以下内容.

客户(app.js)

function connect() {
    var socket = new SockJS('/kafka-data-websocket');
    stompClient = Stomp.over(socket);
    stompClient.connect({}, function (frame) {
        console.log('Connected: ' + frame);
        stompClient.send("/app/fetchData");
        stompClient.subscribe('/data/records', function (message) {
            console.log(JSON.parse(message.body).content);
        });
    });
}

服务器(KafkaController.java)

@Controller
public class KafkaController {

    @MessageMapping("/fetchData")
    @SendTo("/data/records")
    public String fetchMetrics() {
        //...
    }
}

要使用来自特定Kafka主题的数据,我正在使用@KafkaListener注释,如下所示：

public class KafkaReceiver {
    @KafkaListener(topics = "mytopic")
    public void receive(ConsumerRecord<?, ?> record) {
        MyRecord m = new MyRecord(new Long(record.offset()), record.key().toString(), record.value().toString());
           //...
    }
}

我有一个适当的KafkaConfig类,包含所有必需的bean(like explained here).

如何在每个传入/消费的消息上将数据从receive方法发送到KafkaController的fetchMetrics(以及因此到websocket)？

最佳答案 您应该将SimpMessagingTemplate注入KafkaReceiver并从receive()方法中使用它：

 this.template.convertAndSend("/data/records", m);

请参阅Spring Framework Reference Manual中的更多信息.

点赞