Flux OOM实例

2018-02-27 11:51:08来源:oschina作者:go4it人点击

分享

本文主要研究下Flux的OOM产生的场景


FluxSink.OverflowStrategy

reactor-core-3.1.3.RELEASE-sources.jar!/reactor/core/publisher/FluxSink.java


/**
* Enumeration for backpressure handling.
*/
enum OverflowStrategy {
/**
* Completely ignore downstream backpressure requests.
*


* This may yield {@link IllegalStateException} when queues get full downstream.
*/
IGNORE,
/**
* Signal an {@link IllegalStateException} when the downstream can't keep up
*/
ERROR,
/**
* Drop the incoming signal if the downstream is not ready to receive it.
*/
DROP,
/**
* Downstream will get only the latest signals from upstream.
*/
LATEST,
/**
* Buffer all signals if the downstream can't keep up.
*


* Warning! This does unbounded buffering and may lead to {@link OutOfMemoryError}.
*/
BUFFER
}


可以看到BUFFER采用的是无界队列,可能产生OOM



实例
@Test
public void testFluxOOM() throws InterruptedException {
final Flux flux = Flux. create(fluxSink -> {
//NOTE sink:class reactor.core.publisher.FluxCreate$SerializedSink
LOGGER.info("sink:{}",fluxSink.getClass());
while (true) {
fluxSink.next(ThreadLocalRandom.current().nextInt());
}
}, FluxSink.OverflowStrategy.BUFFER)
.publishOn(Schedulers.elastic(),Integer.MAX_VALUE); //NOTE 测试OOM
//NOTE flux:class reactor.core.publisher.FluxCreate,prefetch:-1
LOGGER.info("flux:{},prefetch:{}",flux.getClass(),flux.getPrefetch());
flux.subscribe(e -> {
LOGGER.info("subscribe:{}",e);
try {
TimeUnit.SECONDS.sleep(10);
} catch (InterruptedException e1) {
e1.printStackTrace();
}
});
TimeUnit.MINUTES.sleep(20);
}

jvm参数



-Xmx2160K -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/tmp -XX:+PrintGCDetails -Xloggc:/tmp/gc.log

注意这里使用了publishOn,另外prefetch参数设置为Integer.MAX_VALUE(默认为256),就是为了复现无界队列造成的OOM



输出
java.lang.OutOfMemoryError: GC overhead limit exceeded
Dumping heap to /tmp/java_pid5295.hprof ...
Heap dump file created [6410067 bytes in 0.149 secs]
Exception in thread "main" java.lang.OutOfMemoryError: GC overhead limit exceeded
at java.util.jar.Manifest$FastInputStream.(Manifest.java:332)
at java.util.jar.Manifest$FastInputStream.(Manifest.java:327)
at java.util.jar.Manifest.read(Manifest.java:195)
at java.util.jar.Manifest.(Manifest.java:69)
at java.util.jar.JarFile.getManifestFromReference(JarFile.java:199)
at java.util.jar.JarFile.getManifest(JarFile.java:180)
at sun.misc.URLClassPath$JarLoader$2.getManifest(URLClassPath.java:944)
at java.net.URLClassLoader.defineClass(URLClassLoader.java:450)
at java.net.URLClassLoader.access$100(URLClassLoader.java:73)
Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "main"
Process finished with exit code 137
heap dump

输入图片说明



使用MAT分析可以看到reactor.util.concurrent.SpscLinkedArrayQueue持有了很多未释放的数据,该队列由FluxCreate$BufferAsyncSink持有



static final class BufferAsyncSink extends BaseSink {
final Queue queue;
Throwable error;
volatile boolean done;
volatile int wip;
@SuppressWarnings("rawtypes")
static final AtomicIntegerFieldUpdater WIP =
AtomicIntegerFieldUpdater.newUpdater(BufferAsyncSink.class, "wip");
BufferAsyncSink(CoreSubscriber<? super T> actual, int capacityHint) {
super(actual);
this.queue = Queues.unbounded(capacityHint).get();
}
//......
}

reactor-core-3.1.3.RELEASE-sources.jar!/reactor/util/concurrent/Queues.java


/**
* Returns an unbounded, linked-array-based Queue. Integer.max sized link will
* return the default {@link #SMALL_BUFFER_SIZE} size.
* @param linkSize the link size
* @param the reified {@link Queue} generic type
* @return an unbounded {@link Queue} {@link Supplier}
*/
@SuppressWarnings("unchecked")
public static Supplier> unbounded(int linkSize) {
if (linkSize == XS_BUFFER_SIZE) {
return XS_UNBOUNDED;
}
else if (linkSize == Integer.MAX_VALUE || linkSize == SMALL_BUFFER_SIZE) {
return unbounded();
}
return() -> new SpscLinkedArrayQueue<>(linkSize);
}

可以看到Queues的unbounded方法创建了一个无界队列SpscLinkedArrayQueue来缓冲数据



小结

使用Flux要注意OOM的问题,不过reactor的类库已经尽可能小心地避免这个问题,普通场景的api调用貌似没问题,自己个性化参数的时候要额外注意,本实例就是使用publishOn时特意指定prefetch为Integer.MAX_VALUE,才造成OOM

最新文章

123

最新摄影

闪念基因

微信扫一扫

第七城市微信公众平台