从 Kafka 获取消息，发送到 Rsocket 并从 React 客户端接收答案

【问题标题】：Get Message From Kafka, send to Rsocket and Receive it from React client从 Kafka 获取消息，发送到 Rsocket 并从 React 客户端接收
【发布时间】：2021-04-04 09:19:02
【问题描述】：

我正在尝试使用 Spring 云流将数据从 kafka 发送到 Rsocket，然后在 React 上表示数据

这是我的配置。

@Configuration
public class RsocketConsumerConfiguration {
    
    @Bean
    public Sinks.Many<Data> sender(){
        return Sinks.many().multicast().directBestEffort();
    }
    

}

@控制器公共类服务器控制器 {

@Autowired
private Sinks.Many<Data> integer;

@MessageMapping("integer")
public Flux<Data> integer() {
    return  integer.asFlux();
}

@EnableBinding(IClientProcessor.class)
public class Listener {

    @Autowired
    private Sinks.Many<Data> integer;

    @StreamListener(IClientProcessor.INTEGER)
    public void integer(Data val) {
        System.out.println(val);
        integer.tryEmitNext(val);
    }

}

   let  client = new RSocketClient({
    transport: new RSocketWebSocketClient(
        {
            url: 'ws://localhost:7000/ws',
            wsCreator: (url) => new WebSocket(url),
            debug: true,
        },
        BufferEncoders,
    ),
    setup: {
        dataMimeType: "application/json",
        metadataMimeType: MESSAGE_RSOCKET_COMPOSITE_METADATA.string,
        keepAlive: 5000,
        lifetime: 60000,
    },
});

  client
            .then(rsocket => {
                console.log("Connected to rsocket");
                rsocket.requestStream({
                    metadata: Buffer.from(encodeCompositeMetadata([
                        [MESSAGE_RSOCKET_ROUTING, encodeRoute("integer")],
                    ])),
                 
                })
                    .subscribe({
                        onSubscribe: s => {
                            s.request(2147483647)
                        },
                        onNext: (p) => {
                            let newData = {
                                time: new Date(JSON.parse(p.data).time).getUTCSeconds(),
                                integer: JSON.parse(p.data).integer
                            }
                           newData.integer >100?setInteger(currentData => [newData, ...currentData]):setInt(currentData => [newData, ...currentData])
                           console.log(newData)
                        },
                        onError: (e) => console.error(e),
                        onComplete: () => console.log("Done")
                    });

spring.cloud.stream.bindings.integer.destination=整数无法在反应应用程序中看到它。请指教。我做错了什么？

【问题讨论】：

这个运气好吗？
更新了问题的答案
@meuhedetmeuhedet：不要用解决方案更新问题文本！最好将解决方案发布为您自己问题的答案！您还可以将自己的答案标记为已接受。这样读者就很容易看到问题已经解决了。此外，读者可以对您的答案进行投票。 :)
@Lii 很多新代码。我把它放在哪里？
@meuhedetmeuhedet。我不明白...你不能把代码放在答案文本中的代码块中吗？

标签： reactjs spring spring-boot spring-cloud-stream rsocket

【解决方案1】：

鉴于数据似乎直接从 Kafka（通过 Spring）传输到客户端，stream Kafka messages via an internet-messaging broker to Internet-facing clients over WebSockets 可能更有意义。

披露：我不是那篇文章的作者，但在作者工作的那家公司工作。我们经常看到这种用例，因此希望这种方法可能有用。

【讨论】：