我增強了你的類AsyncTest:
public class RetryableAsyncTest implements Callable<RetryableAsyncTest> {
private final String _name;
private /* */ String _value;
private /* */ boolean _timeouted;
private /* */ int _retryCount;
public RetryableAsyncTest(String name) {
_name = name;
}
@Override
public RetryableAsyncTest call() throws Exception {
try {
++_retryCount;
_timeouted = false;
//-------- Begin of functionnal code
if(Math.random() > 0.5) { // Simulation of
throw new TimeoutException(); // timeout condition
}
_value = "computation result";
//-------- End of functionnal code
}
catch(final TimeoutException x) {
_timeouted = true;
}
return this;
}
public String getName() {
return _name;
}
public String getValue() {
return _value;
}
public boolean isTimeouted() {
return _timeouted;
}
public int getRetryCount() {
return _retryCount;
}
}
RetryableAsyncExecutor類:
public class RetryableAsyncExecutor {
private final ExecutorService _exec;
private final CompletionService<RetryableAsyncTest> _comp;
public RetryableAsyncExecutor(int nThreads) {
_exec = Executors.newFixedThreadPool(nThreads);
_comp = new ExecutorCompletionService<>(_exec);
}
public void submit(RetryableAsyncTest task) {
_comp.submit(task);
}
public RetryableAsyncTest get() throws Exception {
final Future<RetryableAsyncTest> f = _comp.take();
final RetryableAsyncTest task = f.get();
if(task.isTimeouted()) {
_comp.submit(task);
}
return task;
}
public void shutdown() {
_exec.shutdown();
}
}
測試用例:
public class Main {
public static void main(String[] args) {
final int COUNT = 8;
final RetryableAsyncExecutor re = new RetryableAsyncExecutor(5);
try {
for(int i = 0; i < COUNT; ++i) {
re.submit(new RetryableAsyncTest("Async#"+(i+1)));
}
int count = 0;
while(count < COUNT) {
final RetryableAsyncTest task = re.get();
if(task.isTimeouted()) {
System.err.printf("%s: retrying (%d)\n",
task.getName(), task.getRetryCount());
}
else {
System.err.printf("%s: done with '%s'.\n",
task.getName(), task.getValue());
++count;
}
}
}
catch(final Throwable t) {
t.printStackTrace();
}
re.shutdown();
System.exit(0);
}
}
執行日誌:
Async#4: done with 'computation result'.
Async#1: done with 'computation result'.
Async#6: retrying (1)
Async#3: done with 'computation result'.
Async#8: done with 'computation result'.
Async#7: retrying (1)
Async#2: done with 'computation result'.
Async#5: retrying (1)
Async#6: done with 'computation result'.
Async#7: done with 'computation result'.
Async#5: retrying (2)
Async#5: done with 'computation result'.
如果你想小區的重試次數,這樣的邏輯進行到RetryableAsyncExecutor.get()
方法,圍繞_comp.submit(task);
只是一點點澄清之前,我嘗試回答,以確保我知道究竟一個if-then-else條件正在進行。 TimeoutException是來自網絡呼叫滯後,而不是等待執行者完成其業務? – Xype
是的,TimeoutException是來自網絡通話滯後。 – Harsh
而不是雙倍答案,我想你可以從這裏得到你需要的東西https://stackoverflow.com/a/4738630/3630719 – Xype