public static void main(String[] args) {AtomicReference<FluxSink<String>> sendCmdRef = new AtomicReference();Flux<String> sendCmd = Flux.create(aaa -> {System.out.println("create");sendCmdRef.set(aaa);});sendCmd.doOnNext(ttt -> {System.out.println(ttt);}).doOnComplete(() -> {System.out.println("complete");}).subscribe((a) -> {System.out.println(a);});sendCmdRef.get().next("aaaa");sendCmdRef.get().next("bbb");sendCmdRef.get().complete();try {sleep(10000);} catch (InterruptedException e) {e.printStackTrace();}}