java - 多线程应用程序连续失败后如何将主机名添加到阻止列表？

Question

我在代码中使用 Callable ，它将由多个线程调用，如下所示。截至目前，每当抛出任何 RestClientException 时，我都会将 hostname 添加到 blockList。

public class Task implements Callable<DataResponse> {

    private DataKey key;
    private RestTemplate restTemplate;

    public Task(DataKey key, RestTemplate restTemplate) {
        this.key = key;
        this.restTemplate = restTemplate;
    }

    @Override
    public DataResponse call() {
        ResponseEntity<String> response = null;

        // construct what are the hostnames I can call basis on user id
        List<String> hostnames = some_code_here;

        for (String hostname : hostnames) {
            // If host name is null or host name is in block list, skip sending request to this host
            if (DataUtils.isEmpty(hostname) || DataMapping.isBlocked(hostname)) {
                continue;
            }
            try {
                String url = createURL(hostname);
                response = restTemplate.exchange(url, HttpMethod.GET, key.getEntity(), String.class);

                // some code here to return the response if successful
            } catch (HttpClientErrorException ex) {
                // log exception
                return new DataResponse(errorMessage, error, DataStatusEnum.ERROR);
            } catch (HttpServerErrorException ex) {
                // log exception
                return new DataResponse(errorMessage, error, DataStatusEnum.ERROR);
            } catch (RestClientException ex) {
                // I don't want to add it to block list instantly.
                // If same hostname as failed five times consecutively, then only add it
                DataMapping.blockHost(hostname);
            }
        }

        return new DataResponse(DataErrorEnum.SERVER_UNAVAILABLE, DataStatusEnum.ERROR);        
    }
}

下面是我在 DataMapping 类中的内容:

private static final AtomicReference<ConcurrentHashMap<String, String>> blockedHosts = 
        new AtomicReference<ConcurrentHashMap<String, String>>(new ConcurrentHashMap<String, String>());

public static boolean isBlocked(String hostName) {
    return blockedHosts.get().containsKey(hostName);
}

public static void blockHost(String hostName) {
    blockedHosts.get().put(hostName, hostName);
}

问题陈述:-

现在，正如您在 call 方法中看到的那样，一旦 hostname 抛出 RestClientException，我就会阻止它，这可能是不正确的。我需要查看特定的 hostname 是否连续五次抛出 RestClientException，然后仅通过调用此行 将此 hostname 添加到 blockList >DataMapping.blockHost(hostname); 否则不要将其添加到 blockList。

最有效、最好的方法是什么？最多，我总共将拥有 70-100 台独特的机器。

在这种情况下，我的调用方法将从多个线程调用，因此我需要确保为每个主机名正确保留计数，以防它们抛出RestClientException .

编辑:

我在 DataMapping 类中也有以下方法:

我有一个后台线程，每 2 分钟运行一次，它会替换整个集合，因为我的服务提供真实数据，无论是否有任何主机名确实被阻止。我想当我替换整个集合时我确实需要原子引用。

我也在代码中本地添加了阻止功能，因为我可能会在 2 分钟后知道哪台机器被阻止，所以如果可能的话最好提前知道。

// this is being updated from my background thread which runs every 2 minutes
public static void replaceBlockedHosts(List<String> hostNames) {
    ConcurrentHashMap<String, String> newBlockedHosts = new ConcurrentHashMap<>();
    for (String hostName : hostNames) {
        newBlockedHosts.put(hostName, hostName);
    }
    blockedHosts.set(newBlockedHosts);
}

Answer 1

我会将每个主机与一个在每个 RestClientException 上递增的 AtomicInteger 相关联。成功调用后，该整数将设置为零，以强制执行“连续五次”约束。代码看起来像这样。

private final ConcurrentHashMap<String, AtomicInteger> failedCallCount = new ConcurrentHashMap<>();

void call() {
      try {
          String url = createURL(host);
          // make rest call
          resetFailedCallCount(host);
          // ...
      } catch (RestClientException ex) {
          registerFailedCall(host);
          if (shouldBeBlocked(host)) {
              DataMapping.blockHost(host);
          }
      }
}


private boolean shouldBeBlocked(String hostName) {
    AtomicInteger count = failedCallCount.getOrDefault(hostName, new AtomicInteger());
    return count.get() >= 5;
}

private void registerFailedCall(String hostName) {
    AtomicInteger newValue = new AtomicInteger();
    AtomicInteger val = failedCallCount.putIfAbsent(hostName, newValue);
    if (val == null) {
        val = newValue;
    }
    if (val.get() < 5) {
        val.incrementAndGet();
    }
}

private void resetFailedCallCount(String hostName) {
    AtomicInteger count = failedCallCount.get(hostName);
    if (count != null) {
        count.set(0);
    }
}

这是无锁的(至少在我们自己的代码中)并且非常高效。然而，它很容易受到某些竞争条件的影响。最值得注意的是，计数可能会大于 5。但是，这应该不是问题，因为主机无论如何都会被阻止，并且计数不会用于其他任何用途。