javascript - Spring Websocket 与 Kafka 集成

Question

我正在尝试通过 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)每个传入/使用的消息？

Answer 1

您应该将 SimpMessagingTemplate 注入(inject)到 KafkaReceiver 中，并从 receive() 方法中使用它:

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

在 Spring 框架中查看更多信息 Reference Manual .