Как реализовать политику повтора при отправке данных в другое приложение?

Ответ 1

Я не могу разобраться во всех подробностях относительно использования соответствующих API-интерфейсов, но что касается алгоритма, вы можете попробовать:

• политика повтора должна иметь какое-то состояние, прикрепленное к каждому сообщению (по крайней мере, количество повторных попыток текущего сообщения, возможно, что такое текущая задержка). Вам нужно решить, должен ли RetryPolicy хранить это сам или если вы хотите сохранить его внутри сообщения.
• вместо allowRetry, вы могли бы вычислить метод, когда следующая повторная попытка должна произойти (в абсолютном или в миллисекундах в будущем), которая будет функцией вышеупомянутого состояния
• очередь повторов должна содержать информацию о том, когда нужно повторять каждое сообщение.
• вместо scheduleAtFixedRate найдите сообщение в очереди повторных попыток с наименьшим when_is_next_retry (возможно, путем сортировки по абсолютной отметке времени повтора и выбора первого), и пусть служба-исполнитель перенесирует себя с помощью schedule и time_to_next_retry
• для каждого повтора, вытащите его из очереди повтора, отправьте сообщение, используйте RetryPolicy для вычисления, когда следующая повторная попытка должна быть (если она должна быть повторена) и вставьте обратно в очередь повтора с новым значением для when_is_next_retry (если RetryPolicy возвращает -1, это может означать, что сообщение больше не должно быть повторено)

Ответ 2

не идеальный способ, но может быть достигнут и ниже.

public interface RetryPolicy {
public boolean allowRetry();
public void decreaseRetryCount();

}

Создайте две реализации. Для RetryNTimes

public class RetryNTimes implements RetryPolicy {

private int maxRetryCount;
public RetryNTimes(int maxRetryCount) {
    this.maxRetryCount = maxRetryCount;
}

public boolean allowRetry() {
    return maxRetryCount > 0;
}

public void decreaseRetryCount()
{
    maxRetryCount = maxRetryCount-1;
}}

Для ExponentialBackoffRetry

public class ExponentialBackoffRetry implements RetryPolicy {

private int maxRetryCount;
private final Date retryUpto;

public ExponentialBackoffRetry(int maxRetryCount, Date retryUpto) {
    this.maxRetryCount = maxRetryCount;
    this.retryUpto = retryUpto;
}

public boolean allowRetry() {
    Date date = new Date();
    if(maxRetryCount <= 0 || date.compareTo(retryUpto)>=0)
    {
        return false;
    }
    return true;
}

public void decreaseRetryCount() {
    maxRetryCount = maxRetryCount-1;
}}

Вам нужно внести некоторые изменения в класс SendToZeroMQ

public class SendToZeroMQ {

private final ScheduledExecutorService executorService = Executors.newScheduledThreadPool(5);
private final Cache<Long,RetryMessage> retryQueue =
        CacheBuilder
                .newBuilder()
                .maximumSize(10000000)
                .concurrencyLevel(200)
                .removalListener(
                        RemovalListeners.asynchronous(new CustomListener(), executorService)).build();

private static class Holder {
    private static final SendToZeroMQ INSTANCE = new SendToZeroMQ();
}

public static SendToZeroMQ getInstance() {
    return Holder.INSTANCE;
}

private SendToZeroMQ() {
    executorService.submit(new ResponsePoller());
    // retry every 30 seconds for now
    executorService.scheduleAtFixedRate(new Runnable() {
        public void run() {
            for (Map.Entry<Long, RetryMessage> entry : retryQueue.asMap().entrySet()) {
                RetryMessage retryMessage = entry.getValue();
                if(retryMessage.getRetryPolicy().allowRetry())
                {
                    retryMessage.getRetryPolicy().decreaseRetryCount();
                    entry.setValue(retryMessage);
                    sendTo(entry.getKey(), retryMessage.getMessage(),retryMessage);

                }else
                {
                    retryQueue.asMap().remove(entry.getKey());
                }
            }
        }
    }, 0, 30, TimeUnit.SECONDS);
}



public boolean sendTo(final long address, final byte[] encodedRecords, RetryMessage retryMessage) {
    Optional<ZMQSocketInfo> liveSockets = PoolManager.getInstance().getNextSocket();
    if (!liveSockets.isPresent()) {
        return false;
    }
    if(null==retryMessage)
    {
        RetryPolicy retryPolicy = new RetryNTimes(10);
        retryMessage = new RetryMessage(retryPolicy,encodedRecords);
        retryQueue.asMap().put(address,retryMessage);
    }
    return sendTo(address, encodedRecords, liveSockets.get().getSocket());
}

public boolean sendTo(final long address, final byte[] encodedByteArray, final ZMQ.Socket socket) {
    ZMsg msg = new ZMsg();
    msg.add(encodedByteArray);
    boolean sent = msg.send(socket);
    msg.destroy();
    return sent;
}

public void removeFromRetryQueue(final long address) {
    retryQueue.invalidate(address);
}}

Ответ 3

Вы можете использовать apache camel. Он обеспечивает компонент для zeromq, и инструменты, такие как errohandler, redeliverypolicy, канал с неверной информацией и такие вещи, предоставляются в основном.

Ответ 4

Вот небольшая симуляция вашей среды, которая показывает, как это можно сделать. Обратите внимание, что кеш Гуавы здесь неправильная структура данных, так как вы не заинтересованы в выселении (я думаю). Поэтому я использую одновременный hashmap:

package experimental;

import static java.util.concurrent.TimeUnit.MILLISECONDS;

import java.util.Arrays;
import java.util.Iterator;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;

class Experimental {
  /** Return the desired backoff delay in millis for the given retry number, which is 1-based. */
  interface RetryStrategy {
    long getDelayMs(int retry);
  }

  enum ConstantBackoff implements RetryStrategy {
    INSTANCE;
    @Override
    public long getDelayMs(int retry) {
      return 1000L;
    }
  }

  enum ExponentialBackoff implements RetryStrategy {
    INSTANCE;
    @Override
    public long getDelayMs(int retry) {
      return 100 + (1L << retry);
    }
  }

  static class Sender {
    private final ScheduledExecutorService executorService = Executors.newScheduledThreadPool(4);
    private final ConcurrentMap<Long, Retrier> pending = new ConcurrentHashMap<>();

    /** Send the given data with given address on the given socket. */
    void sendTo(long addr, byte[] data, int socket) {
      System.err.println("Sending " + Arrays.toString(data) + "@" + addr + " on " + socket);
    }

    private class Retrier implements Runnable {
      private final RetryStrategy retryStrategy;
      private final long addr;
      private final byte[] data;
      private final int socket;
      private int retry;
      private Future<?> future; 

      Retrier(RetryStrategy retryStrategy, long addr, byte[] data, int socket) {
        this.retryStrategy = retryStrategy;
        this.addr = addr;
        this.data = data;
        this.socket = socket;
        this.retry = 0;
      }

      synchronized void start() {
        if (future == null) {
          future = executorService.submit(this);
          pending.put(addr, this);
        }
      }

      synchronized void cancel() {
        if (future != null) {
          future.cancel(true);
          future = null;
        }
      }

      private synchronized void reschedule() {
        if (future != null) {
          future = executorService.schedule(this, retryStrategy.getDelayMs(++retry), MILLISECONDS);
        }
      }

      @Override
      synchronized public void run() {
        sendTo(addr, data, socket);
        reschedule();
      }
    }

    long getVerifiedAddr() {
      System.err.println("Pending messages: " + pending.size());
      Iterator<Long> i = pending.keySet().iterator();
      long addr = i.hasNext() ? i.next() : 0;
      return addr;
    }

    class CancellationPoller implements Runnable {
      @Override
      public void run() {
        while (!Thread.currentThread().isInterrupted()) {
          try {
            Thread.sleep(1000);
          } catch (InterruptedException ex) { 
            Thread.currentThread().interrupt();
          }
          long addr = getVerifiedAddr();
          if (addr == 0) {
            continue;
          }
          System.err.println("Verified message (to be cancelled) " + addr);
          Retrier retrier = pending.remove(addr);
          if (retrier != null) {
            retrier.cancel();
          }
        }
      }
    }

    Sender initialize() {
      executorService.submit(new CancellationPoller());
      return this;
    }

    void sendWithRetriesTo(RetryStrategy retryStrategy, long addr, byte[] data, int socket) {
      new Retrier(retryStrategy, addr, data, socket).start();
    }
  }

  public static void main(String[] args) {
    Sender sender = new Sender().initialize();
    for (long i = 1; i <= 10; i++) {
      sender.sendWithRetriesTo(ConstantBackoff.INSTANCE, i, null, 42);
    }
    for (long i = -1; i >= -10; i--) {
      sender.sendWithRetriesTo(ExponentialBackoff.INSTANCE, i, null, 37);
    }
  }
}