5

我为 RSocket 消息写了一个小演示

问题是我无法访问Rsocket端点，我从服务器收到以下异常：

客户端： 配置：

@Bean
RSocket rSocket() {
    return RSocketFactory.connect()
            .mimeType(MimeTypeUtils.APPLICATION_JSON_VALUE, MimeTypeUtils.APPLICATION_JSON_VALUE)
            .frameDecoder(PayloadDecoder.ZERO_COPY)
            .transport(TcpClientTransport.create(new InetSocketAddress(7500)))
            .start()
            .block();
}

@Bean
RSocketRequester requester(RSocketStrategies strategies) {
    return RSocketRequester.wrap(rSocket(), MimeTypeUtils.APPLICATION_JSON, MimeTypeUtils.APPLICATION_JSON, strategies);
}

控制器：

private final RSocketRequester requester;

@GetMapping("/greet/{name}")
public Publisher<GreetingsResponse> greet(@PathVariable String name) {
    return requester
            .route("hello")
            .data(new GreetingsRequest(name))
            .retrieveMono(GreetingsResponse.class);
}

服务器端（使用spring Rsocket）： yml：

spring:
  rsocket:
    server:
      port: 7500
      transport: tcp
  main:
    lazy-initialization: true

配置：

@MessageMapping("hello")
Mono<GreetingsResponse> greet(GreetingsRequest request) {
    return Mono.just(new GreetingsResponse("Hello " + request.getName() + " @ " + Instant.now()));
}

我很确定它与新wrap函数有关，RSocketRequester.wrap 因为它接受一个新参数metadataMimeType，我将它设置为 application/Json，但它似乎不起作用

堆栈跟踪：

org.springframework.messaging.MessageDeliveryException：在 org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler.handleNoMatch(RSocketMessageHandler.java:312) 在 org.springframework.messaging.handler.invocation.reactive 中没有目标处理程序。 AbstractMethodMessageHandler.getHandlerMethod(AbstractMethodMessageHandler.java:445) at org.springframework.messaging.handler.invocation.reactive.AbstractMethodMessageHandler.handleMessage(AbstractMethodMessageHandler.java:417) at org.springframework.messaging.rsocket.annotation.support.MessagingRSocket.lambda$ handleAndReply$4(MessagingRSocket.java:173) 在 reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:44) 在 reactor.core.publisher.Mono.subscribe(Mono.java:3920) 在 reactor.core.publisher。FluxConcatArray$ConcatArraySubscriber.onComplete(FluxConcatArray.java:207) 在 reactor.core.publisher.FluxConcatArray.subscribe(FluxConcatArray.java:80) 在 reactor.core.publisher.MonoFromFluxOperator.subscribe(MonoFromFluxOperator.java:74) 在 io.rsocket .RSocketResponder.handleRequestResponse(RSocketResponder.java:386) 在 reactor.core.publisher.LambdaSubscriber.onNext(LambdaSubscriber.java:160) 在 reactor.core.publisher.RSocketResponder.handleFrame(RSocketResponder.java:298)。 MonoFlatMapMany$FlatMapManyInner.onNext(MonoFlatMapMany.java:238) 在 reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.drainRegular(FluxGroupBy.java:554) 在 reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.drain(FluxGroupBy.java:630)在 reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux。在 reactor.core.publisher.MonoFlatMapMany$FlatMapManyMain.onNext(MonoFlatMapMany.java:184) 在 reactor.core.publisher 订阅(FluxGroupBy.java:696).Flux.subscribe(Flux.java:8000) 在 reactor.core.publisher .Operators$MonoSubscriber.complete(Operators.java:1582) 在 reactor.core.publisher.MonoProcessor.onNext(MonoProcessor.java:316) 在 io.rsocket.internal.ClientServerInputMultiplexer.lambda$new$1(ClientServerInputMultiplexer.java:116)在 reactor.core.publisher.LambdaSubscriber.onNext(LambdaSubscriber.java:160) 在 reactor.core.publisher.FluxGroupBy$GroupByMain.drainLoop(FluxGroupBy.java:380) 在 reactor.core.publisher.FluxGroupBy$GroupByMain.drain(FluxGroupBy .java:316) 在 reactor.core.publisher.FluxGroupBy$GroupByMain.onNext(FluxGroupBy.java:201) 在 reactor.core.publisher.FluxMap$MapSubscriber。onNext(FluxMap.java:114) 在 reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:114) 在 reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:206) 在 reactor.netty.channel .FluxReceive.onInboundNext(FluxReceive.java:322) 在 reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:342) 在 reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:91) 在 io.netty。 channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360) at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352) at io.netty .handler.codec.ByteToMessageDecoder。fireChannelRead(ByteToMessageDecoder.java:328) at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:302) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374) at io.netty.channel .AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360) 在 io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352) 在 io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1421) 在 io。 netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374) 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360) 在 io.netty.channel。DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:930) at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:163) at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:697)在 io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:632) 在 io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:549) 在 io.netty.channel.nio.NioEventLoop.run (NioEventLoop.java:511) 在 io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918) 在 io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) 在 io。 netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) 在 java.base/java.lang.Thread.run(Thread.java:834)fireChannelRead(DefaultChannelPipeline.java:930) at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:163) at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:697) at io .netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:632) at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:549) at io.netty.channel.nio.NioEventLoop.run(NioEventLoop .java:511) 在 io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) 在 io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918) 在 io.netty。 util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) 在 java.base/java.lang.Thread.run(Thread.java:834)fireChannelRead(DefaultChannelPipeline.java:930) at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:163) at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:697) at io .netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:632) at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:549) at io.netty.channel.nio.NioEventLoop.run(NioEventLoop .java:511) 在 io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) 在 io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918) 在 io.netty。 util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) 在 java.base/java.lang.Thread.run(Thread.java:834)930) 在 io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:697) 在 io.netty.channel.nio 的 io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:163) .NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:632) 在 io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:549) 在 io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:511) 在io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918) 在 io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) 在 io.netty.util.concurrent.FastThreadLocalRunnable。在 java.base/java.lang.Thread.run(Thread.java:834) 处运行（FastThreadLocalRunnable.java:30）930) 在 io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:697) 在 io.netty.channel.nio 的 io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:163) .NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:632) 在 io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:549) 在 io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:511) 在io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918) 在 io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) 在 io.netty.util.concurrent.FastThreadLocalRunnable。在 java.base/java.lang.Thread.run(Thread.java:834) 处运行（FastThreadLocalRunnable.java:30）在 io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:632) 在 io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:632) 在 io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:697) 读取(AbstractNioByteChannel.java:163) .channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:549) at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:511) at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor. java:918) 在 io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) 在 io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) 在 java.base/java。 lang.Thread.run(Thread.java:834)在 io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:632) 在 io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:632) 在 io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:697) 读取(AbstractNioByteChannel.java:163) .channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:549) at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:511) at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor. java:918) 在 io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) 在 io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) 在 java.base/java。 lang.Thread.run(Thread.java:834)processSelectedKeysOptimized(NioEventLoop.java:632) at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:549) at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:511) at io.netty .util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918) at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable .java:30) 在 java.base/java.lang.Thread.run(Thread.java:834)processSelectedKeysOptimized(NioEventLoop.java:632) at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:549) at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:511) at io.netty .util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918) at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable .java:30) 在 java.base/java.lang.Thread.run(Thread.java:834)netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) at java.base/java.lang.Thread.run(Thread .java:834)netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) at java.base/java.lang.Thread.run(Thread .java:834)

4

2 回答 2

8

您使用的是哪个春季版本？我遇到了同样的问题，我通过更改 spring-boot-starter-parent 2.2.0.M3 解决了这个问题。

这是我的来源 https://github.com/han1448/spring-rsocket-example

添加。

我解决了这个问题。您需要将 mimeType 更改为message/x.rsocket.routing.v0. 您可以从MetadataExtractor.ROUTING.

@Bean
RSocket rSocket() {
    return RSocketFactory.connect()
            .mimeType(MetadataExtractor.ROUTING.toString(), MimeTypeUtils.APPLICATION_JSON_VALUE)
            .frameDecoder(PayloadDecoder.ZERO_COPY)
            .transport(TcpClientTransport.create(new InetSocketAddress(7500)))
            .start()
            .block();
}

@Bean
RSocketRequester requester(RSocketStrategies strategies) {
    return RSocketRequester.wrap(rSocket(), MimeTypeUtils.APPLICATION_JSON, MetadataExtractor.ROUTING, strategies);
}
于 2019-08-10T08:21:56.360 回答
4

对于RSocketRequester，方法名从create到wrap，然后添加第三个参数。所以，我觉得现在还不稳定。

我的选择是使用 builder 方法，如下所示。

@Bean
RSocketRequester rSocketRequester(RSocketStrategies strategies) {
    InetSocketAddress address = new InetSocketAddress(clientConfigProp.getHost(), clientConfigProp.getPort());
    log.info("RSocket server address={}", address);
    return RSocketRequester.builder()
        .rsocketFactory(factory -> factory
            .dataMimeType(MimeTypeUtils.ALL_VALUE)
            .frameDecoder(PayloadDecoder.ZERO_COPY))
        .rsocketStrategies(strategies)
        .connect(TcpClientTransport.create(address))
        .retry().block();
}
于 2019-09-14T17:47:11.047 回答