[case7]Flux OOM實例

序

本文主要研究下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

使用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