RxJava中：如何在onComplete回调时安全取消Flowable订阅？
2024-12-13 18:16:03
0浏览
收藏
从现在开始，努力学习吧！本文《RxJava中：如何在onComplete回调时安全取消Flowable订阅？》主要讲解了等等相关知识点，我会在golang学习网中持续更新相关的系列文章，欢迎大家关注并积极留言建议。下面就先一起来看一下本篇正文内容吧，希望能帮到你！
在 rxjava 中，如何在 oncomplete 被调用时取消订阅？
通过在 oncomplete 回调中使用 completablefuture 可以实现取消订阅。
在提供的代码示例中，flowable 首先被转换为一个 table 列表，每个表包含表的详细信息和列。然后，该列表被转换为一个 flowable，逐个发出每个表。
要取消订阅，可在 oncomplete 回调中使用 completablefuture.complete 方法，通知调用方操作已完成。调用方可以轮询 completablefuture，直到它完成，然后调用 dispose() 方法取消订阅。
以下是修改后的代码示例：
consoleschemaflowableoutput.java
...
@override
public disposable flush(information information, flowable
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
tableFlowable = result.getAll(dbName.get(), strategy).flatMap(new Function>() {
@Override
public Publisher apply(@NonNull Table table) throws Exception {
return result.getTableColumn(table).flatMap(new Function, SingleSource>() {
@Override
public SingleSource
中国自动驾驶初创公司小马智行在美国IPO融资2.6亿美元，估值45.5亿美元 
