在 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("\r\n");
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 被调用时取消订阅，从而释放资源并终止流处理。
