如何在将数据发送到另一个应用程序时实施重试策略？

Question

我正在处理将数据发送到的应用程序zeromq.以下是我的应用程序:

• 我有一个类SendToZeroMQ将数据发送到zeromq.
• 将相同的数据添加到retryQueue同一个类中,以便以后可以在未收到确认的情况下重试.它使用带有maximumSize限制的guava缓存.
• 有一个单独的线程从zeromq接收先前发送的数据的确认,如果没有收到确认,SendToZeroMQ则将重试发送相同的数据.如果收到确认,那么我们将把它删除,retryQueue以便不能再次重试.

想法很简单,我必须确保我的重试策略正常工作,这样我就不会丢失我的数据.这是非常罕见的,但如果我们没有收到acknolwedgements.

我正在考虑构建两种类型,RetryPolicies但我无法理解如何在这里构建与我的程序相对应的:

• RetryNTimes: 在此,它将在每次重试之间以特定的睡眠重试N次,之后,它将丢弃记录.
• ExponentialBackoffRetry:在此,它将指数级地继续重试.我们可以设置一些最大重试限制,然后它将不会重试并将丢弃记录.

下面是我的SendToZeroMQ类,它将数据发送到zeromq,也从后台线程每30秒重试一次并启动ResponsePollerrunnable,它会一直运行:

public class SendToZeroMQ {
  private final ScheduledExecutorService executorService = Executors.newScheduledThreadPool(5);
  private final Cache<Long, byte[]> 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() {
      @Override
      public void run() {
        for (Entry<Long, byte[]> entry : retryQueue.asMap().entrySet()) {
          sendTo(entry.getKey(), entry.getValue());
        }
      }
    }, 0, 30, TimeUnit.SECONDS);
  }

  public boolean sendTo(final long address, final byte[] encodedRecords) {
    Optional<ZMQSocketInfo> liveSockets = PoolManager.getInstance().getNextSocket();
    if (!liveSockets.isPresent()) {
      return false;
    }
    return sendTo(address, encodedRecords, liveSockets.get().getSocket());
  }

  public boolean sendTo(final long address, final byte[] encodedByteArray, final Socket socket) {
    ZMsg msg = new ZMsg();
    msg.add(encodedByteArray);
    boolean sent = msg.send(socket);
    msg.destroy();
    // adding to retry queue
    retryQueue.put(address, encodedByteArray);
    return sent;
  }

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

下面是我的ResponsePoller类,它会轮询zeromq的所有确认.如果我们从zeromq收到确认,那么我们将从重试队列中删除该记录,以便它不会被重试,否则它将被重试.

public class ResponsePoller implements Runnable {
  private static final Random random = new Random();

  @Override
  public void run() {
    ZContext ctx = new ZContext();
    Socket client = ctx.createSocket(ZMQ.PULL);
    String identity = String.format("%04X-%04X", random.nextInt(), random.nextInt());
    client.setIdentity(identity.getBytes(ZMQ.CHARSET));
    client.bind("tcp://" + TestUtils.getIpaddress() + ":8076");

    PollItem[] items = new PollItem[] {new PollItem(client, Poller.POLLIN)};

    while (!Thread.currentThread().isInterrupted()) {
      // Tick once per second, pulling in arriving messages
      for (int centitick = 0; centitick < 100; centitick++) {
        ZMQ.poll(items, 10);
        if (items[0].isReadable()) {
          ZMsg msg = ZMsg.recvMsg(client);
          Iterator<ZFrame> it = msg.iterator();
          while (it.hasNext()) {
            ZFrame frame = it.next();
            try {
                long address = TestUtils.getAddress(frame.getData());
                // remove from retry queue since we got the acknowledgment for this record
                SendToZeroMQ.getInstance().removeFromRetryQueue(address);               
            } catch (Exception ex) {
                // log error
            } finally {
              frame.destroy();
            }
          }
          msg.destroy();
        }
      }
    }
    ctx.destroy();
  }
}

---

题:

正如您在上面所看到的,我encodedRecords使用SendToZeroMQ类发送给zeromq ,然后每30秒重试一次,具体取决于我们是否从ResponsePoller课堂上获得了一个acknolwedgement .

对于每一个encodedRecords都有一个唯一的密钥被调用address,这就是我们将从zeromq获得的确认.

我如何继续并扩展此示例以构建我上面提到的两个重试策略,然后我可以选择在发送数据时要使用的重试策略.我提出了以下界面,但后来我无法理解我应该如何继续实施这些重试策略并在上面的代码中使用它.

public interface RetryPolicy {
    /**
     * Called when an operation has failed for some reason. This method should return
     * true to make another attempt.
     */
    public boolean allowRetry(int retryCount, long elapsedTimeMs);
}

我可以在这里使用guava-retrying或failsafe,因为这些库已经有许多我可以使用的重试策略吗？

Answer 1

我无法计算出有关如何使用相关API的所有细节,但对于算法,您可以尝试:

• 重试策略需要在每条消息上附加某种状态(至少重试当前消息的次数,可能是当前的延迟).您需要确定RetryPolicy是应该保留它自己还是要将其存储在消息中.
• 而不是allowRetry,你可以有一个方法计算下一次重试应该发生的时间(绝对时间或未来的毫秒数),这将是上述状态的函数
• 重试队列应包含有关何时应重试每条消息的信息.
• 而不是使用scheduleAtFixedRate,在重试队列中找到具有最低的消息when_is_next_retry(可能通过对绝对重试时间戳进行排序并选择第一个),并让executorService使用schedule和重新安排自己.time_to_next_retry
• 对于每次重试,将其从重试队列中拉出,发送消息,使用RetryPolicy计算下次重试的时间(如果要重试)并使用新值插入重试队列when_is_next_retry(如果RetryPolicy返回-1,这可能意味着不再重试该消息)