在 rxJava 中，如何在 oncomplete 被调用时取消订阅？
通过在 oncomplete 回调中使用 completablefuture 可以实现取消订阅。
在提供的代码示例中，flowable 首先被转换为一个 table 列表，每个表包含表的详细信息和列。然后，该列表被转换为一个 flowable，逐个发出每个表。
要取消订阅，可在 oncomplete 回调中使用 completablefuture.complete 方法，通知调用方操作已完成。调用方可以轮询 completablefuture，直到它完成，然后调用 dispose() 方法取消订阅。
立即学习“Java免费学习笔记（深入）”；
以下是修改后的代码示例：
consoleschemaflowableoutput.java
... @override public disposable flush(information information, flowable<table> table) throws schemaexportexception { logger.info("start flowable flush"); disposable export_flush_complete = table.subscribe(tableins -> { system.out.println(printasciitable(tableins)); system.out.println(printasciicolumns(tableins.getcolumns())); system.out.println(" "); count.addandget(1); }, throwable -> { logger.debug("export break, reason: " + throwable.getmessage()); future.cancel(true); throw new schemaexportexception(throwable); }, new action() { @override public void run() throws exception { logger.debug("export complete, affect size:" + count.get()); future.complete("ok"); } }); return export_flush_complete; } ...
调用方
... long startStamp = System.currentTimeMillis(); // Flowable Flowable<Table> tableFlowable = result.getAll(dbName.get(), strategy).flatMap(new Function<Table, Publisher<Table>>() { @Override public Publisher<Table> apply(@NonNull Table table) throws Exception { return result.getTableColumn(table).flatMap(new Function<List<Column>, SingleSource<Table>>() { @Override public SingleSource<Table> apply(@NonNull List<Column> columns) throws Exception { return Single.just(table.fillColumn(columns)); } }).flatMapPublisher(new Function<Table, Publisher<? extends Table>>() { @Override public Publisher<? extends Table> apply(@NonNull Table table) throws Exception { return Flowable.just(table); } }); } }); Disposable disposable = out.flush(info, tableFlowable); ...
通过使用 completablefuture，调用方可以在 oncomplete 被调用时取消订阅，从而释放资源并终止流处理。
