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

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类中的addAndSendJunkedclose方法返回布尔值,那么我需要依赖哪个布尔值?因为这两种方法中的任何一种都可以发送数据。

  • 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机制的原理