使用一个策略之一将记录发送到消息队列

Posted

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了使用一个策略之一将记录发送到消息队列相关的知识，希望对你有一定的参考价值。

我有一堆密钥(clientKey)和值(processBytes)我想通过将它们打包到一个字节数组中发送到我们的消息队列。我将创建所有键和值的一个字节数组，该数组应始终小于50K，然后发送到我们的消息队列。

对于每个分区，我有一堆dataHolders所以我正在迭代这些，然后将它发送到我的消息队列： - 

private void validateAndSend(final DataPartition partition) {
  final ConcurrentLinkedQueue<DataHolder> dataHolders = dataHoldersByPartition.get(partition);

  // sending data via async policy but it can be send with other two sync queue policy as well.
  final Packet packet = new Packet(partition, new QPolicyAsync());

  DataHolder dataHolder;
  while ((dataHolder = dataHolders.poll()) != null) {
    packet.addAndSendJunked(dataHolder.getClientKey().getBytes(StandardCharsets.UTF_8),
        dataHolder.getProcessBytes());
  }
  packet.close();
}

Packet类：此类将所有键和值打包到一个字节数组中，并调用构造函数中传递的相应实现以将数据发送到队列。

public final class Packet implements Closeable {
  private static final int MAX_SIZE = 50000;
  private static final int HEADER_SIZE = 36;

  private final byte dataCenter;
  private final byte recordVersion;
  private final long address;
  private final long addressFrom;
  private final long addressOrigin;
  private final byte partition;
  private final byte replicated;
  private final ByteBuffer itemBuffer = ByteBuffer.allocate(MAX_SIZE);
  private final QueuePolicy policy;
  private int pendingItems = 0;

  public Packet(final DataPartition partition, final QueuePolicy policy) {
    this.partition = (byte) partition.getPartition();
    this.policy = policy;
    this.dataCenter = Utils.LOCATION.get().datacenter();
    this.recordVersion = 1;
    this.replicated = 0;
    final long packedAddress = new Data().packAddress();
    this.address = packedAddress;
    this.addressFrom = 0L;
    this.addressOrigin = packedAddress;
  }

  private void addHeader(final ByteBuffer buffer, final int items) {
    buffer.put(dataCenter).put(recordVersion).putInt(items).putInt(buffer.capacity())
        .putLong(address).putLong(addressFrom).putLong(addressOrigin).put(partition)
        .put(replicated);
  }

  // sending here by calling policy implementation
  private void sendData() {
    if (itemBuffer.position() == 0) {
      // no data to be sent
      return;
    }
    final ByteBuffer buffer = ByteBuffer.allocate(MAX_SIZE);
    addHeader(buffer, pendingItems);
    buffer.put(itemBuffer);
    // sending data via particular policy
    policy.sendToQueue(address, buffer.array());
    itemBuffer.clear();
    pendingItems = 0;
  }

  public void addAndSendJunked(final byte[] key, final byte[] data) {
    if (key.length > 255) {
      return;
    }
    final byte keyLength = (byte) key.length;
    final byte dataLength = (byte) data.length;

    final int additionalSize = dataLength + keyLength + 1 + 1 + 8 + 2;
    final int newSize = itemBuffer.position() + additionalSize;
    if (newSize >= (MAX_SIZE - HEADER_SIZE)) {
      sendData();
    }
    if (additionalSize > (MAX_SIZE - HEADER_SIZE)) {
      throw new AppConfigurationException("Size of single item exceeds maximum size");
    }

    final ByteBuffer dataBuffer = ByteBuffer.wrap(data);
    final long timestamp = dataLength > 10 ? dataBuffer.getLong(2) : System.currentTimeMillis();
    // data layout
    itemBuffer.put((byte) 0).put(keyLength).put(key).putLong(timestamp).putShort(dataLength)
        .put(data);
    pendingItems++;
  }

  @Override
  public void close() {
    if (pendingItems > 0) {
      sendData();
    }
  }
}

现在，我可以通过三种不同的方式将数据发送到我的消息传递队列，因此我创建了一个接口，然后有三种不同的实现：

QueuePolicy界面：

public interface QueuePolicy {
    public boolean sendToQueue(final long address, final byte[] encodedRecords);
}

QPolicyAsync类：

public class QPolicyAsync implements QueuePolicy {

  @Override
  public boolean sendToQueue(long address, byte[] encodedRecords) {
    return SendRecord.getInstance().sendToQueueAsync(address, encodedRecords);
  }
}

QPolicySync类：

public class QPolicySync implements QueuePolicy {

  @Override
  public boolean sendToQueue(long address, byte[] encodedRecords) {
    return SendRecord.getInstance().sendToQueueSync(address, encodedRecords);
  }
}

QPolicySyncWithSocket类：

public class QPolicySyncWithSocket implements QueuePolicy {
  private final Socket socket;

  public QPolicySyncWithSocket(Socket socket) {
    this.socket = socket;
  }

  @Override
  public boolean sendToQueue(long address, byte[] encodedRecords) {
    return SendRecord.getInstance().sendToQueueSync(address, encodedRecords, Optional.of(socket));
  }
}

这个想法很简单：我通过这三个QueuePolicy实现中的任何一个将数据发送到我的消息队列。这取决于客户端如何发送数据。截至目前，我正在QueuePolicy构造函数中传递Packet的实现，然后通过该策略发送数据。每个QueuePolicy实现调用SendRecord类中的相应方法。

现在我需要知道数据是否成功发送。截至目前，Packet类中的方法不返回任何布尔值，因此我不知道它是否成功发送。我可以遇到dataHolders中只有一个元素或者它可以包含多个元素的情况。

private void validateAndSend(final DataPartition partition) {
  final ConcurrentLinkedQueue<DataHolder> dataHolders = dataHoldersByPartition.get(partition);

  // sending data via async policy but it can be send with other two sync queue policy as well.
  final Packet packet = new Packet(partition, new QPolicyAsync());

  DataHolder dataHolder;
  while ((dataHolder = dataHolders.poll()) != null) {
    packet.addAndSendJunked(dataHolder.getClientKey().getBytes(StandardCharsets.UTF_8),
        dataHolder.getProcessBytes());
  }
  packet.close();
  // how do I know whether this data was successfully sent?

}

如果我从Packet类中的addAndSendJunked和close方法返回布尔值，那么我需要依赖哪个布尔值？因为这两种方法中的任何一种都可以发送数据。

  • close方法将发送数据，其中只有一个元素或剩下的元素。
  • addAndSendJunked方法将在达到限制时立即发送数据。

更新：

以下是我更新的代码：

public final class Packet implements Closeable {
  private static final int MAX_SIZE = 50000;
  private static final int HEADER_SIZE = 36;
  private boolean result = false;

  private final byte dataCenter;
  private final byte recordVersion;
  private final long address;
  private final long addressFrom;
  private final long addressOrigin;
  private final byte partition;
  private final byte replicated;
  private final ByteBuffer itemBuffer = ByteBuffer.allocate(MAX_SIZE);
  private final QueuePolicy policy;
  private int pendingItems = 0;

  public Packet(final DataPartition partition, final QueuePolicy policy) {
    this.partition = (byte) partition.getPartition();
    this.policy = policy;
    this.dataCenter = Utils.LOCATION.get().datacenter();
    this.recordVersion = 1;
    this.replicated = 0;
    final long packedAddress = new Data().packAddress();
    this.address = packedAddress;
    this.addressFrom = 0L;
    this.addressOrigin = packedAddress;
  }

  private void addHeader(final ByteBuffer buffer, final int items) {
    buffer.put(dataCenter).put(recordVersion).putInt(items).putInt(buffer.capacity())
        .putLong(address).putLong(addressFrom).putLong(addressOrigin).put(partition)
        .put(replicated);
  }

  // sending here by calling policy implementation
  private void flush() {
    if (itemBuffer.position() == 0) {
      // no data to be sent
      return true;
    }
    final ByteBuffer buffer = ByteBuffer.allocate(MAX_SIZE);
    addHeader(buffer, pendingItems);
    buffer.put(itemBuffer);
    // sending data via particular policy
    boolean sent = policy.sendToQueue(address, buffer.array());
    itemBuffer.clear();
    pendingItems = 0;
    return sent;
  }

  public void addAndSendJunked(final byte[] key, final byte[] data) {
    if (key.length > 255) {
      result = false;
      return;
    }
    final byte keyLength = (byte) key.length;
    final byte dataLength = (byte) data.length;

    final int additionalSize = dataLength + keyLength + 1 + 1 + 8 + 2;
    final int newSize = itemBuffer.position() + additionalSize;
    if (newSize >= (MAX_SIZE - HEADER_SIZE)) {
      result = flush();
    }
    if (additionalSize > (MAX_SIZE - HEADER_SIZE)) {
      throw new AppConfigurationException("Size of single item exceeds maximum size");
    }

    final ByteBuffer dataBuffer = ByteBuffer.wrap(data);
    final long timestamp = dataLength > 10 ? dataBuffer.getLong(2) : System.currentTimeMillis();
    // data layout
    itemBuffer.put((byte) 0).put(keyLength).put(key).putLong(timestamp).putShort(dataLength)
        .put(data);
    pendingItems++;
  }

  @Override
  public void close() {
    if (pendingItems > 0) {
      result = flush();
    }
  }

  public boolean getResult() {
    return result;
  }
}
答案

你不能从close()方法返回一个布尔值，因为它被覆盖了。

你有不同的选择：

  1. 每当send返回false时抛出异常
  2. 从sendData返回一个布尔值，我将重命名为flush并将其公开（见下文），并从addAndSendJunked返回一个布尔值。
  3. 在类Packet中添加一个布尔字段和一个getter，以便能够随时获取其值

方法flush：

public boolean flush() {
    if (itemBuffer.position() == 0) {
        // no data to be sent
        return true;
    }
    final ByteBuffer buffer = ByteBuffer.allocate(MAX_SIZE);
    addHeader(buffer, pendingItems);
    buffer.put(itemBuffer);
    // sending data via particular policy
    boolean result = policy.sendToQueue(address, buffer.array());
    itemBuffer.clear();
    pendingItems = 0;
    return result;
}

以上是关于使用一个策略之一将记录发送到消息队列的主要内容，如果未能解决你的问题，请参考以下文章

IPC System V 消息队列 - 发送一个数组块

Linux C语言高级编程之使用消息队列实现进程间通信！重点内容！！！

RabbitMQ之发布订阅

mysql之消息队列

Symfony 信使队列与绑定键 - 重试策略

handle机制的原理