问题描述
我将 ReactiveMongoRepository 与 Kotlin 一起使用，但我不时收到此内存泄漏错误：
堆栈跟踪：
LEAK: ByteBuf.release() was not called before it's garbage-collected. See https://netty.io/wiki/reference-counted-objects.html for more information. Recent access records:
Created at: io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:385)
io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:187)
io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:178)
io.netty.buffer.AbstractByteBufAllocator.buffer(AbstractByteBufAllocator.java:115)
org.springframework.core.io.buffer.NettyDataBufferFactory.allocateBuffer(NettyDataBufferFactory.java:71)
org.springframework.core.io.buffer.NettyDataBufferFactory.allocateBuffer(NettyDataBufferFactory.java:39)
org.springframework.http.codec.json.AbstractJackson2Encoder.encodeValue(AbstractJackson2Encoder.java:236)
org.springframework.http.codec.ServerSentEventHttpMessageWriter.encodeEvent(ServerSentEventHttpMessageWriter.java:174)
org.springframework.http.codec.ServerSentEventHttpMessageWriter.lambda$encode$0(ServerSentEventHttpMessageWriter.java:160)
reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:106)
reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:120)
reactor.core.publisher.FluxUsingWhen$UsingWhenSubscriber.onNext(FluxUsingWhen.java:358)
reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:79)
reactor.core.publisher.MonoFlatMapMany$FlatMapManyInner.onNext(MonoFlatMapMany.java:250)
reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.innerNext(FluxConcatMap.java:281)
reactor.core.publisher.FluxConcatMap$ConcatMapInner.onNext(FluxConcatMap.java:860)
reactor.core.publisher.FluxConcatMap$WeakScalarsubscription.request(FluxConcatMap.java:476)
reactor.core.publisher.Operators$MultiSubscriptionSubscriber.set(Operators.java:2193)
reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.drain(FluxConcatMap.java:442)
reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.onNext(FluxConcatMap.java:250)
com.mongodb.reactivestreams.client.internal.AbstractSubscription.onNext(AbstractSubscription.java:155)
com.mongodb.reactivestreams.client.internal.AbstractSubscription.processResultsQueue(AbstractSubscription.java:230)
com.mongodb.reactivestreams.client.internal.AbstractSubscription.tryProcessResultsQueue(AbstractSubscription.java:185)
com.mongodb.reactivestreams.client.internal.MongoIterableSubscription.lambda$requestMoreData$1(MongoIterableSubscription.java:99)
com.mongodb.internal.async.ErrorHandlingResultCallback.onResult(ErrorHandlingResultCallback.java:48)
com.mongodb.internal.operation.AsyncQueryBatchCursor.handleGetMoreQueryResult(AsyncQueryBatchCursor.java:351)
com.mongodb.internal.operation.AsyncQueryBatchCursor.access$600(AsyncQueryBatchCursor.java:56)
com.mongodb.internal.operation.AsyncQueryBatchCursor$CommandResultSingleResultCallback.onResult(AsyncQueryBatchCursor.java:383)
com.mongodb.internal.operation.AsyncQueryBatchCursor$CommandResultSingleResultCallback.onResult(AsyncQueryBatchCursor.java:356)
com.mongodb.internal.async.ErrorHandlingResultCallback.onResult(ErrorHandlingResultCallback.java:48)
com.mongodb.internal.connection.DefaultServer$DefaultServerProtocolExecutor$2.onResult(DefaultServer.java:288)
com.mongodb.internal.async.ErrorHandlingResultCallback.onResult(ErrorHandlingResultCallback.java:48)
com.mongodb.internal.connection.CommandProtocolImpl$1.onResult(CommandProtocolImpl.java:84)
com.mongodb.internal.connection.DefaultConnectionPool$PooledConnection$2.onResult(DefaultConnectionPool.java:530)
com.mongodb.internal.connection.UsageTrackingInternalConnection$2.onResult(UsageTrackingInternalConnection.java:142) c
om.mongodb.internal.async.ErrorHandlingResultCallback.onResult(ErrorHandlingResultCallback.java:48)
com.mongodb.internal.connection.InternalStreamConnection$2$1.onResult(InternalStreamConnection.java:463)
com.mongodb.internal.connection.InternalStreamConnection$2$1.onResult(InternalStreamConnection.java:440)
com.mongodb.internal.connection.InternalStreamConnection$MessageHeaderCallback$MessageCallback.onResult(InternalStreamConnection.java:745)
com.mongodb.internal.connection.InternalStreamConnection$MessageHeaderCallback$MessageCallback.onResult(InternalStreamConnection.java:712)
com.mongodb.internal.connection.InternalStreamConnection$5.completed(InternalStreamConnection.java:582)
com.mongodb.internal.connection.InternalStreamConnection$5.completed(InternalStreamConnection.java:579)
com.mongodb.connection.netty.NettyStream.readAsync(NettyStream.java:255)
com.mongodb.connection.netty.NettyStream.readAsync(NettyStream.java:214)
com.mongodb.internal.connection.InternalStreamConnection.readAsync(InternalStreamConnection.java:579)
com.mongodb.internal.connection.InternalStreamConnection.access$1100(InternalStreamConnection.java:78)
com.mongodb.internal.connection.InternalStreamConnection$MessageHeaderCallback.onResult(InternalStreamConnection.java:702)
com.mongodb.internal.connection.InternalStreamConnection$MessageHeaderCallback.onResult(InternalStreamConnection.java:687)
com.mongodb.internal.connection.InternalStreamConnection$5.completed(InternalStreamConnection.java:582)
com.mongodb.internal.connection.InternalStreamConnection$5.completed(InternalStreamConnection.java:579)
com.mongodb.connection.netty.NettyStream.readAsync(NettyStream.java:255)
com.mongodb.connection.netty.NettyStream.readAsync(NettyStream.java:214)
com.mongodb.connection.netty.NettyStream.handleReadResponse(NettyStream.java:285)
com.mongodb.connection.netty.NettyStream.access$800(NettyStream.java:69)
com.mongodb.connection.netty.NettyStream$InboundBufferHandler.channelRead0(NettyStream.java:344)
com.mongodb.connection.netty.NettyStream$InboundBufferHandler.channelRead0(NettyStream.java:341)
io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1533)
io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1282)
io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1329)
io.netty.handler.codec.BytetoMessageDecoder.decodeRemovalReentryProtection(BytetoMessageDecoder.java:508)
io.netty.handler.codec.BytetoMessageDecoder.callDecode(BytetoMessageDecoder.java:447)
io.netty.handler.codec.BytetoMessageDecoder.channelRead(BytetoMessageDecoder.java:276)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:719)
io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:655)
io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:581)
io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493)
io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
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)
我不知道如何解决这个问题。
一些代码：
// REST
@GetMapping("/global",produces = [MediaType.TEXT_EVENT_STREAM_VALUE])
fun getAll(): Flux<PersonDto> {
return personService.findAllCompanyIdNull()
}
// Service
fun findAllCompanyIdNull(): Flux<PersonDto> = personRepository.findAllByCompanyIdisNull().map { it.toDto() }
// Repo
interface PersonRepository : ReactiveMongoRepository<Person,ObjectId> {
fun findAllByCompanyId(companyId: Long): Flux<Person>
}
这个错误在我的服务器中不时发生。
Callback onResult call produced an error com.mongodb.MongoException: Subscription has already been terminated at com.mongodb.reactivestreams.client.internal.AbstractSubscription.onError
我的代码非常简单并且标准地使用了 API，知道如何解决这个问题吗？
解决方法
暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！
如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。
小编邮箱:dio#foxmail.com (将#修改为@）