Я использую ниже класс для отправки данных в нашу очередь обмена сообщениями, используя сокет либо синхронно, либо асинхронно, как показано ниже.
-
sendAsync- Он отправляет данные асинхронно без какого-либо таймаута. После отправки(on LINE A)он добавляет в ведроretryHolderтак, что, если подтверждение не получено, оно снова повторится из фонового потока, который запускается в конструкторе. -
send- он внутренне вызывает методsendAsync, а затем выполняет спящий режим для определенного периода ожидания и, если подтверждение не получено, оно удаляется из ведраretryHolder, чтобы мы не повторили попытку.
Таким образом, единственное различие между этими двумя вышеописанными методами: - Для async мне нужно выполнить любую попытку, но для синхронизации мне не нужно повторять попытку, но похоже, что это может быть повторено, поскольку мы используем один и тот же кеш кеша повтора и поток повторений выполняется каждые 1 секунду.
ResponsePoller - это класс, который получает подтверждение для данных, которые были отправлены в нашу очередь сообщений, а затем вызывает метод removeFromretryHolder ниже, чтобы удалить адрес, чтобы мы не повторили попытку после получения подтверждения.
public class SendToQueue {
private final ExecutorService cleanupExecutor = Executors.newFixedThreadPool(5);
private final ScheduledExecutorService executorService = Executors.newScheduledThreadPool(3);
private final Cache<Long, byte[]> retryHolder =
CacheBuilder
.newBuilder()
.maximumSize(1000000)
.concurrencyLevel(100)
.removalListener(
RemovalListeners.asynchronous(new LoggingRemovalListener(), cleanupExecutor)).build();
private static class Holder {
private static final SendToQueue INSTANCE = new SendToQueue();
}
public static SendToQueue getInstance() {
return Holder.INSTANCE;
}
private SendToQueue() {
executorService.submit(new ResponsePoller()); // another thread which receives acknowledgement and then delete entry from the `retryHolder` cache accordingly.
executorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
// retry again
for (Entry<Long, byte[]> entry : retryHolder.asMap().entrySet()) {
sendAsync(entry.getKey(), entry.getValue());
}
}
}, 0, 1, TimeUnit.SECONDS);
}
public boolean sendAsync(final long address, final byte[] encodedRecords, final Socket socket) {
ZMsg msg = new ZMsg();
msg.add(encodedRecords);
// send data on a socket LINE A
boolean sent = msg.send(socket);
msg.destroy();
retryHolder.put(address, encodedRecords);
return sent;
}
public boolean send(final long address, final byte[] encodedRecords, final Socket socket) {
boolean sent = sendAsync(address, encodedRecords, socket);
// if the record was sent successfully, then only sleep for timeout period
if (sent) {
try {
TimeUnit.MILLISECONDS.sleep(500);
} catch (InterruptedException ex) {
Thread.currentThread().interrupt();
}
}
// if key is not present, then acknowledgement was received successfully
sent = !retryHolder.asMap().containsKey(address);
// and key is still present in the cache, then it means acknowledgment was not received after
// waiting for timeout period, so we will remove it from cache.
if (!sent)
removeFromretryHolder(address);
return sent;
}
public void removeFromretryHolder(final long address) {
retryHolder.invalidate(address);
}
}
Каков наилучший способ, с помощью которого мы не пытаемся повторить попытку, если кто-либо вызывает метод send, но нам все равно нужно знать, было ли получено подтверждение или нет. Единственное, что мне не нужно вообще повторять.
Нужна ли нам отдельная ведро для всех вызовов синхронизации только для подтверждения, и мы не возвращаемся из этого ведра?
