序

本文主要研究下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.
         * <p>
         * 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.
         * <p>
         * Warning! This does unbounded buffering and may lead to {@link OutOfMemoryError}.
         */
        BUFFER
    }

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

实例

    @Test
    public void testFluxOOM() throws InterruptedException {
        final Flux<Integer> flux = Flux.<Integer> 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.<init>(Manifest.java:332)
    at java.util.jar.Manifest$FastInputStream.<init>(Manifest.java:327)
    at java.util.jar.Manifest.read(Manifest.java:195)
    at java.util.jar.Manifest.<init>(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

flux-oom.png

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

    static final class BufferAsyncSink<T> extends BaseSink<T> {

        final Queue<T> queue;

        Throwable error;
        volatile boolean done;

        volatile int wip;
        @SuppressWarnings("rawtypes")
        static final AtomicIntegerFieldUpdater<BufferAsyncSink> WIP =
                AtomicIntegerFieldUpdater.newUpdater(BufferAsyncSink.class, "wip");

        BufferAsyncSink(CoreSubscriber<? super T> actual, int capacityHint) {
            super(actual);
            this.queue = Queues.<T>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 <T> the reified {@link Queue} generic type
     * @return an unbounded {@link Queue} {@link Supplier}
     */
    @SuppressWarnings("unchecked")
    public static <T> Supplier<Queue<T>> 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