Spring Boot中RSocket客户端如何正确添加Bearer Token
问题背景
基于Java + Spring Boot开发RSocket客户端时,在客户端侧添加Bearer Token的环节出现异常,服务端执行RSocket Security校验时抛出访问拒绝错误,请求链路最终断开。
客户端原有实现代码
Flux<RSocket> clientRSocket = RSocketConnector.create() .metadataMimeType(WellKnownMimeType.MESSAGE_RSOCKET_COMPOSITE_METADATA.getString()) .dataMimeType(WellKnownMimeType.APPLICATION_JSON.getString()) .keepAlive(Duration.ofMinutes(10), Duration.ofMinutes(10)).payloadDecoder(PayloadDecoder.ZERO_COPY) .connect(transport).flux(); CompositeByteBuf metadata = ByteBufAllocator.DEFAULT.compositeBuffer(); RoutingMetadata routingMetadata = TaggingMetadataCodec.createRoutingMetadata(ByteBufAllocator.DEFAULT, List.of("/test")); ByteBuf simpleAuthentication = AuthMetadataCodec.encodeBearerMetadata(ByteBufAllocator.DEFAULT, TenantContext.getCurrentToken().toCharArray()); CompositeMetadataCodec.encodeAndAddMetadata(metadata, ByteBufAllocator.DEFAULT, WellKnownMimeType.MESSAGE_RSOCKET_ROUTING, routingMetadata.getContent()); CompositeMetadataCodec..encodeAndAddMetadata(metadata, ByteBufAllocator.DEFAULT, WellKnownMimeType.MESSAGE_RSOCKET_AUTHENTICATION, simpleAuthentication.asByteBuf()); clientRSocket.subscribe(rsoc -> { Disposable requestStream = rsoc.requestStream(DefaultPayload.create(data, metadata)) .subscribeOn(Schedulers.newBoundedElastic(1, 100, UUID.randomUUID().toString())).map(mapper -> { return mapper.getDataUtf8(); }).subscribe(); });
服务端报错堆栈
org.springframework.security.access.AccessDeniedException: Access Denied at org.springframework.security.authorization.ReactiveAuthorizationManager.lambda$verify$1(ReactiveAuthorizationManager.java:53) at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:44) at reactor.core.publisher.Mono.subscribe(Mono.java:4210) at reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onComplete(FluxSwitchIfEmpty.java:75) at reactor.core.publisher.FluxFilter$FilterSubscriber.onComplete(FluxFilter.java:160) at reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onComplete(FluxSwitchIfEmpty.java:78) at reactor.core.publisher.MonoNext$NextSubscriber.onComplete(MonoNext.java:96) at reactor.core.publisher.MonoNext$NextSubscriber.onNext(MonoNext.java:77) at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.innerNext(FluxConcatMap.java:274) at reactor.core.publisher.FluxConcatMap$ConcatMapInner.onNext(FluxConcatMap.java:851) at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1712) at reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:241) at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1712) at reactor.core.publisher.FluxDefaultIfEmpty$DefaultIfEmptySubscriber.onComplete(FluxDefaultIfEmpty.java:100) at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:136) at reactor.core.publisher.FluxFilter$FilterSubscriber.onComplete(FluxFilter.java:160) at reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onComplete(FluxSwitchIfEmpty.java:78) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onComplete(FluxMapFuseable.java:144) at reactor.core.publisher.FluxFilterFuseable$FilterFuseableSubscriber.onComplete(FluxFilterFuseable.java:165) at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1713) at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:144) at reactor.core.publisher.FluxFilterFuseable$FilterFuseableSubscriber.onNext(FluxFilterFuseable.java:112) at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:2274) at reactor.core.publisher.FluxFilterFuseable$FilterFuseableSubscriber.request(FluxFilterFuseable.java:184) at reactor.core.publisher.MonoFlatMap$FlatMapMain.onSubscribe(MonoFlatMap.java:103) at reactor.core.publisher.FluxFilterFuseable$FilterFuseableSubscriber.onSubscribe(FluxFilterFuseable.java:81) at reactor.core.publisher.MonoCurrentContext.subscribe(MonoCurrentContext.java:35) at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:55) at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:150) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:121) at reactor.core.publisher.FluxFilterFuseable$FilterFuseableSubscriber.onNext(FluxFilterFuseable.java:112) at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:2274) at reactor.core.publisher.FluxFilterFuseable$FilterFuseableSubscriber.request(FluxFilterFuseable.java:184) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:162) at reactor.core.publisher.MonoFlatMap$FlatMapMain.onSubscribe(MonoFlatMap.java:103) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90) at reactor.core.publisher.FluxFilterFuseable$FilterFuseableSubscriber.onSubscribe(FluxFilterFuseable.java:81) at reactor.core.publisher.MonoJust.subscribe(MonoJust.java:54) at reactor.core.publisher.Mono.subscribe(Mono.java:4210) at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.drain(FluxConcatMap.java:441) at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.onSubscribe(FluxConcatMap.java:211) at reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:161) at reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:86) at reactor.core.publisher.Mono.subscribe(Mono.java:4210) at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.drain(MonoIgnoreThen.java:172) at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:56) at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:150) at reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67) at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1712) at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:144) at reactor.core.publisher.FluxFilterFuseable$FilterFuseableSubscriber.onNext(FluxFilterFuseable.java:112) at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:2274) at reactor.core.publisher.FluxFilterFuseable$FilterFuseableSubscriber.request(FluxFilterFuseable.java:184) at reactor.core.publisher.MonoFlatMap$FlatMapMain.onSubscribe(MonoFlatMap.java:103) at reactor.core.publisher.FluxFilterFuseable$FilterFuseableSubscriber.onSubscribe(FluxFilterFuseable.java:81) at reactor.core.publisher.MonoCurrentContext.subscribe(MonoCurrentContext.java:35) at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:55) at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) at reactor.core.publisher.Mono.subscribe(Mono.java:4210) at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.drain(MonoIgnoreThen.java:172) at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:56) at reactor.core.publisher.Mono.subscribe(Mono.java:4210) at reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onComplete(FluxSwitchIfEmpty.java:75) at reactor.core.publisher.Operators.complete(Operators.java:135) at reactor.core.publisher.MonoEmpty.subscribe(MonoEmpty.java:45) at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:55) at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) at reactor.core.publisher.Mono.subscribe(Mono.java:4210) at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.drain(MonoIgnoreThen.java:172) at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:56) at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:55) at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) at reactor.core.publisher.Flux.subscribe(Flux.java:8264) at io.rsocket.internal.RateLimitableRequestPublisher.subscribe(RateLimitableRequestPublisher.java:74) at io.rsocket.RSocketResponder.handleStream(RSocketResponder.java:446) at io.rsocket.RSocketResponder.handleFrame(RSocketResponder.java:310) at reactor.core.publisher.LambdaSubscriber.onNext(LambdaSubscriber.java:160) at reactor.core.publisher.MonoFlatMapMany$FlatMapManyInner.onNext(MonoFlatMapMany.java:242) at reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.drainRegular(FluxGroupBy.java:554) at reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.drain(FluxGroupBy.java:630) at reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.subscribe(FluxGroupBy.java:696) at reactor.core.publisher.Flux.subscribe(Flux.java:8264) at reactor.core.publisher.MonoFlatMapMany$FlatMapManyMain.onNext(MonoFlatMapMany.java:188) at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1712) at reactor.core.publisher.MonoProcessor.onNext(MonoProcessor.java:317) at io.rsocket.internal.ClientServerInputMultiplexer.lambda$new$1(ClientServerInputMultiplexer.java:116) at reactor.core.publisher.LambdaSubscriber.onNext(LambdaSubscriber.java:160) at reactor.core.publisher.FluxGroupBy$GroupByMain.drainLoop(FluxGroupBy.java:380) at reactor.core.publisher.FluxGroupBy$GroupByMain.drain(FluxGroupBy.java:316) at reactor.core.publisher.FluxGroupBy$GroupByMain.onNext(FluxGroupBy.java:201) at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:114) at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:114) at reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:220) at reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:353) at reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:352) at reactor.netty.http.server.HttpServerOperations.onInboundNext(HttpServerOperations.java:493) at reactor.netty.http.server.WebsocketServerOperations.onInboundNext(WebsocketServerOperations.java:154) at reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:90) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365) at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357) at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:321) at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:295) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365) at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357) at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365) at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919) at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:163) at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:714) at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:650) at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:576) at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493) at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989) at io.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:832) java.nio.channels.ClosedChannelException
堆栈末尾的
java.nio.channels.ClosedChannelException是权限校验失败后服务端主动断开连接导致的连带异常,不需要单独排查,解决权限校验问题后该异常会自动消失。
问题根因与修复方案
原有代码存在4个直接触发异常的问题,逐一修复即可解决:
- 语法笔误:写入认证元数据的代码行
CompositeMetadataCodec..encodeAndAddMetadata多写了一个点,属于基础语法错误,会导致元数据写入逻辑执行异常,先修正为单点的标准方法调用。 - 认证元数据重复包装:
AuthMetadataCodec.encodeBearerMetadata方法返回的ByteBuf本身已经按照RSocket认证协议规范封装了认证类型、Token长度等帧头信息,原有代码额外调用.asByteBuf()会导致内容重复包装,服务端解析时无法识别到合法的Bearer Token,这是直接触发AccessDenied异常的核心原因。 - 认证位置错误:Spring Security RSocket默认在连接建立的SETUP帧阶段校验身份信息,原有代码只把Token放在单个业务请求的元数据里,连接阶段服务端读不到任何认证信息,会直接判定为未授权访问。
- 内存泄漏风险:手动创建的CompositeByteBuf、Payload没有配置释放逻辑,长时间运行会触发Netty内存泄漏,间接导致连接异常断开。
修复后的可正常运行代码示例:
// 配置RSocket连接器,在连接建立阶段携带Bearer Token RSocketConnector connector = RSocketConnector.create() .metadataMimeType(WellKnownMimeType.MESSAGE_RSOCKET_COMPOSITE_METADATA.getString()) .dataMimeType(WellKnownMimeType.APPLICATION_JSON.getString()) .keepAlive(Duration.ofMinutes(10), Duration.ofMinutes(10)) .payloadDecoder(PayloadDecoder.ZERO_COPY) // SETUP帧写入认证元数据,供服务端连接阶段校验 .setupPayload((setupMetadata, allocator) -> { CompositeByteBuf connectionMetadata = allocator.compositeBuffer(); // 直接传入编码后的Bearer Token,不需要额外做ByteBuf包装 ByteBuf bearerTokenBuf = AuthMetadataCodec.encodeBearerMetadata(allocator, TenantContext.getCurrentToken().toCharArray()); CompositeMetadataCodec.encodeAndAddMetadata(connectionMetadata, allocator, WellKnownMimeType.MESSAGE_RSOCKET_AUTHENTICATION, bearerTokenBuf); return PayloadUtils.createPayload(connectionMetadata); }); // 发起业务请求 connector.connect(transport) .flatMapMany(rsocket -> { CompositeByteBuf reqMetadata = ByteBufAllocator.DEFAULT.compositeBuffer();
相关产品推荐
相关产品推荐

