序

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