4

当我RSocketRequester这样创建时它工作正常：

@Bean
RSocketRequester rSocketRequester(RSocketStrategies rSocketStrategies) {
    return RSocketRequester.builder()
            .rsocketStrategies(rSocketStrategies)
            .connectTcp("localhost", 7000)
            .block();
}

但是当我尝试发送消息时它会抛出一个异常，如果这样：

@Bean
RSocket rSocket() {
    return RSocketFactory
            .connect()
            .transport(TcpClientTransport.create(7000))
            .start()
            .block();
}

@Bean
RSocketRequester rSocketRequester(RSocketStrategies rSocketStrategies) {
    return RSocketRequester.wrap(rSocket(), MediaType.APPLICATION_CBOR, MimeTypeUtils.parseMimeType(WellKnownMimeType.MESSAGE_RSOCKET_COMPOSITE_METADATA.getString()), rSocketStrategies);
}

io.rsocket.exceptions.ApplicationErrorException：在 io.rsocket.exceptions.Exceptions.from(Exceptions.java:45) ~[rsocket-core-1.0.0-RC5.jar:na] 处没有目标“处理程序”已抑制：reactor .core.publisher.FluxOnAssembly$OnAssemblyException：在以下站点观察到错误：|_ checkpoint ⇢ 处理程序 com.issoft.rnd.ms2.controller.TestController#test4() [DispatcherHandler] |_ checkpoint ⇢ HTTP GET “/test4” [ExceptionHandlingWebHandler] 堆栈跟踪：在 io.rsocket.exceptions.Exceptions.from(Exceptions.java:45) ~[rsocket-core-1.0.0-RC5.jar:na] 在 io.rsocket.RSocketRequester。 handleFrame(RSocketRequester.java:556) ~[rsocket-core-1.0.0-RC5.jar:na] at io.rsocket.RSocketRequester.handleIncomingFrames(RSocketRequester.java:516) ~[rsocket-core-1.0.0-RC5 。罐：na] 在 reactor.core.publisher.LambdaSubscriber.onNext(LambdaSubscriber.java:160) ~[reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE] 在 reactor.core.publisher.MonoFlatMapMany$FlatMapManyInner。 onNext(MonoFlatMapMany.java:242) ~[reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE] at reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.drainRegular(FluxGroupBy.java:554) ~[reactor -core-3.3.0.RELEASE.jar:3.3.0.RELEASE] 在 reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.drain(FluxGroupBy.java:630) ~[reactor-core-3.3.0.RELEASE.jar: 3.3.0.RELEASE] 在 reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.subscribe(FluxGroupBy.java:696) ~[reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE] 在 reactor.core。 publisher.Flux.subscribe(Flux.java:8134) ~[reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE] 在 reactor.core.publisher.MonoFlatMapMany$FlatMapManyMain.onNext(MonoFlatMapMany.java:188) ~[reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE] 在 reactor.core.publisher.Operators$ MonoSubscriber.complete(Operators.java:1592) ~[reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE] 在 reactor.core.publisher.MonoProcessor.onNext(MonoProcessor.java:317) ~[reactor -core-3.3.0.RELEASE.jar:3.3.0.RELEASE] 在 io.rsocket.internal.ClientServerInputMultiplexer.lambda$new$1(ClientServerInputMultiplexer.java:116) ~[rsocket-core-1.0.0-RC5.jar :na] 在 reactor.core.publisher.LambdaSubscriber.onNext(LambdaSubscriber.java:160) ~[reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE] 在 reactor.core.publisher.FluxGroupBy$GroupByMain .drainLoop(FluxGroupBy.java:380) ~[reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE] 在 reactor.core.publisher.FluxGroupBy$GroupByMain.drain(FluxGroupBy.java:316) ~[reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE] 在 reactor.core.publisher.FluxGroupBy$ GroupByMain.onNext(FluxGroupBy.java:201) ~[reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE] at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:114) ~ [reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE] 在 reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:114) ~[reactor-core-3.3.0.RELEASE. jar:3.3.0.RELEASE] 在 reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:213) ~[reactor-netty-0.9.1.RELEASE.jar:0.9.1.RELEASE] 在 reactor.netty。 channel.FluxReceive.onInboundNext(FluxReceive.java:346) ~[reactor-netty-0.9.1.RELEASE.jar:0.9.1.RELEASE] 在 reactor.netty.channel。ChannelOperations.onInboundNext(ChannelOperations.java:348) ~[reactor-netty-0.9.1.RELEASE.jar:0.9.1.RELEASE] 在 reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:93) ~[reactor -netty-0.9.1.RELEASE.jar:0.9.1.RELEASE] 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374) ~[netty-transport-4.1.43.Final.jar:4.1. 43.Final] 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360) ~[netty-transport-4.1.43.Final.jar:4.1.43.Final] 在 io.netty.channel.AbstractChannelHandlerContext。 fireChannelRead(AbstractChannelHandlerContext.java:352) ~[netty-transport-4.1.43.Final.jar:4.1.43.Final] at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:326) ~[netty-codec-4.1.43.Final.jar:4.1.43.Final] at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:300) ~[netty-codec-4.1 .43.Final.jar:4.1.43.Final] 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374) ~[netty-transport-4.1.43.Final.jar:4.1.43.Final]在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360) ~[netty-transport-4.1.43.Final.jar:4.1.43.Final] 在 io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext. java:352) ~[netty-transport-4.1.43.Final.jar:4.1.43.Final] at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1422) ~[netty-transport-4.1 .43.Final.jar:4.1.43。Final] at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374) ~[netty-transport-4.1.43.Final.jar:4.1.43.Final] at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead( AbstractChannelHandlerContext.java:360) ~[netty-transport-4.1.43.Final.jar:4.1.43.Final] at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:931) ~[netty-transport-4.1 .43.Final.jar:4.1.43.Final] 在 io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:163) ~[netty-transport-4.1.43.Final.jar:4.1. 43.Final] 在 io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:700) ~[netty-transport-4.1.43.Final.jar:4.1.43.Final] 在 io.netty.channel。 nio.NioEventLoop。processSelectedKeysOptimized(NioEventLoop.java:635) ~[netty-transport-4.1.43.Final.jar:4.1.43.Final] at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:552) ~[netty -transport-4.1.43.Final.jar:4.1.43.Final] 在 io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:514) ~[netty-transport-4.1.43.Final.jar: 4.1.43.Final] 在 io.netty.util.concurrent.SingleThreadEventExecutor$6.run(SingleThreadEventExecutor.java:1050) ~[netty-common-4.1.43.Final.jar:4.1.43.Final] 在 io.netty .util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) ~[netty-common-4.1.43.Final.jar:4.1.43.Final] at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable .java:30) ~[netty-common-4.1.43.Final.jar:4.1.43.Final] 在 java.base/java.lang.Thread.run(Thread.java:834) ~[na:呐]

发送功能：

public Mono<String> request(String route, String data, Class<String> clazz) {
    return rSocketRequester.route(route).data(data).retrieveMono(clazz);
}

服务器：

@Controller
@AllArgsConstructor
public class RTestController {

    private EchoService echoService;

    @MessageMapping("test")
    Mono<String> test(String value) {
        return echoService.echo(value);
    }
}

Spring Boot 版本： 2.2.1.RELEASE

注意： Mime 类型取自第一个工作版本。

功能有什么问题wrap？

4

2 回答 2

4

TL；DR：这不是 Spring 的问题，而是您创建客户端 RSocket 连接的方式。

该RSocketRequester.wrap方法通常在服务器端使用，一旦客户端发起连接并且服务器想要向另一方发送请求。Javadoc 说您可以在客户端使用它，这是一个完全有效的用例。

现在RSocketRequester.wrap，在其参数中接受有关预期数据 mime 类型和元数据 mime 类型的数据。这通常在建立连接时在两方之间协商，使用 RSocket SETUP/METADATA_PUSH 帧。一旦建立连接，它就完成了。该wrap方法需要这些参数，因为我们无法从现有的 RSocket 中轻松提取这些信息。

在您的情况下，连接已经创建，而RSocketFactory没有提供该信息（ClientRSocketFactory该类为此提供了许多方法）。所以我认为元数据无法在请求中正确编码，因此服务器无法读取路由信息。

除非您有非常具体的要求，否则我建议您使用RSocketRequester.BuilderAPI 创建客户端 RSocket，因为这样更容易正确处理。

于 2019-11-15T08:35:26.977 回答
2

添加dataMimeType和metadataMimeType帮助。

@Bean
RSocket rSocket() {
    return RSocketFactory
            .connect()
            .dataMimeType(MediaType.APPLICATION_CBOR_VALUE)
            .metadataMimeType(WellKnownMimeType.MESSAGE_RSOCKET_COMPOSITE_METADATA.getString())
            .transport(TcpClientTransport.create(7000))
            .start()
            .block();
}

使用 wrap 方法的最初原因是使用 LoadBalancedRSocketMono，它提供客户端负载平衡和重新连接 OOTB。

LoadBalancedRSocketMono.create(Flux.just(Collections.singleton(new RSocketSupplier(() ->
                        RSocketFactory
                                .connect()
                                .dataMimeType(MediaType.APPLICATION_CBOR_VALUE)
                                .metadataMimeType(WellKnownMimeType.MESSAGE_RSOCKET_COMPOSITE_METADATA.getString())
                                .frameDecoder(PayloadDecoder.ZERO_COPY)
                                .transport(TcpClientTransport.create("localhost", 7000))
                                .start()
                                .doOnNext(rSocket -> {
                                    this.rSocketRequester = RSocketRequester.wrap(rSocket, MediaType.APPLICATION_CBOR, MimeTypeUtils.parseMimeType(WellKnownMimeType.MESSAGE_RSOCKET_COMPOSITE_METADATA.getString()), rSocketStrategies);
                                })
                )))
        ).subscribe();
于 2019-11-18T07:17:03.667 回答