我有一个 http 客户端和执行器,当所有工作完成后应该将其关闭。
我正在尝试按照此处针对 RxJava 1.x 描述的方式使用 Flux.using 方法: https://github.com/meddle0x53/learning-rxjava/blob/master/src/main/java/com/packtpub/reactive/chapter08/ResourceManagement.java
我的资源创建方法:
public static Flux<GithubClient> createResource(String token,
int connectionCount) {
return Flux.using(
() -> {
logger.info(Thread.currentThread().getName() + " : Created and started the client.");
return new GithubClient(token, connectionCount);
},
client -> {
logger.info(Thread.currentThread().getName() + " : About to create Observable.");
return Flux.just(client);
},
client -> {
logger.info(Thread.currentThread().getName() + " : Closing the client.");
client.close();
},
false
).doOnSubscribe(subscription -> logger.info("subscribed"));
}
然后我使用:
Flux<StateMutator> dataMutators = GithubClient.createResource(
config.getAccessToken(),
config.getConnectionCount())
.flatMap(client -> client.loadRepository(organization, repository)
问题是客户端连接在发出第一个请求之前就已关闭。
[main] INFO com.sapho.services.githubpublic.client.GithubClient - main : Created and started the client.
[main] INFO com.sapho.services.githubpublic.client.GithubClient - main : About to create Observable.
[main] INFO com.sapho.services.githubpublic.client.GithubClient - subscribed
[main] INFO com.sapho.services.githubpublic.client.GithubClient - main : Closing the client.
java.lang.IllegalStateException: Client instance has been closed.
at jersey.repackaged.com.google.common.base.Preconditions.checkState(Preconditions.java:173)
at org.glassfish.jersey.client.JerseyClient.checkNotClosed(JerseyClient.java:273)
没有找到任何 Reactor 的示例。
谢谢
最佳答案
我再次阅读了使用文档,发现了我的错误。使用 return Flux.just(client);
返回客户端没有意义,因为 Flux 会立即终止,从而触发客户端关闭。
我最终实现了:
public static Flux<StateMutator> createAndExecute(GithubPublicConfiguration config,
Function<GithubClient, Flux<StateMutator>> toExecute) {
return Flux.using(
() -> {
logger.debug(Thread.currentThread().getName() + " : Created and started the client.");
return new GithubClient(entityModelHandler, config.getAccessToken(), config.getConnectionCount());
},
client -> toExecute.apply(client),
client -> {
logger.debug(Thread.currentThread().getName() + " : Closing the client.");
client.close();
},
false
);
}
然后我调用它:
GithubClient.createAndExecute(config,
client -> client.loadRepository(organization, repository))
现在所有操作都按适当的顺序进行。
关于java - 如何正确管理 Reactor 中的可关闭资源,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/42037552/