如何阻止现有的异步网络请求？

Question

我已经向第三方库提供了回调，该库在不同时间调用提供的方法，为我提供已更改的对象。然后我执行异步 Web 请求以获取更多详细信息并将它们设置在该对象上，下面是一个类似的示例；

public void update(Person person) {
     if (person.getId() == -1) {
         mService.getPersonDetails()
                 .flatMap(..)
                 .skip(..)
                 .subscribe(personResult -> person.setId(personResult.getId()))
     }
}

更新被调用了很多次，应该只在对象没有 ID 时执行查询。问题是至少有两个请求被发送，因为第一个查询尚未完成。

如何同步此方法调用，以便为通过回调传递的每个对象只发送一个请求？我只想阻止对该确切对象的请求，因此如果 update() 提供不同的对象，则可以发送新请求。

Answer 1

好吧，你可以简单地 synchronize 它。

public synchronized void update(Person person)

Answer 2

您可以使用 distinct operator. Here's a general idea of how you could do that using a PublishSubject (JavaDoc 过滤您的 observable 的输入）（注意这是凭记忆写的，我还没有测试过）：

private PublishSubject<Person> personSubject;
public void update(Person person) {
     if (personSubject == null) {
         personSubject = new PublishSubject();
         personSubject
            .filter(person -> person.getId() == -1)
            .distinct()
            .flatMap(person -> mService.getPersonDetails())
            .skip(..)
            .subscribe(personResult -> person.setId(personResult.getId()));
     }
     personSubject.onNext(person);
}

当然，您必须在 Person class 上实现 equals 方法（正如 Marek 指出的那样，这将导致传入的所有对象被缓存在内存中）或实现 distinct(Func) 变体。

该方法采用 'key selector' 函数来区分对象。如果您的对象相当重并且您担心内存（例如，如果您在 Android 上），这可能是更好的方法。像这样：

.distinct(new Func1<Person, Integer>() {
                @Override
                public Integer call(Person person) {
                    return person.hashCode();
                }
            })

Answer 3

Adam S提供的解决方案看起来不错，但迟早会出现OOM问题。这是由于不同的运算符必须存储所有唯一值。

我想到的其他选项是使用 ConcurrentMap 来存储已处理的人员并使用 doOnTerminate 来清理它。

    private Map<Person, Boolean> map = new ConcurrentHashMap<>();
    public void update(final Person person) {
     if (person.getId() == -1) {
        if(map.putIfAbsent(person, true)==null){
           mService.getPersonDetails()
                 .flatMap(..)
                 .skip(..)
                 .doOnTerminate(()->map.remove(person))
                 .subscribe(personResult -> person.setId(personResult.getId()))
        }
     }
}

如何阻止现有的异步网络请求？

How to block existing async web requests?

java

rx-java