У меня есть набор ключей и значений, которые я хочу отправить в нашу очередь обмена сообщениями, упаковав их в один байтовый массив. Я сделаю однобайтовый массив всех ключей и значений, который всегда должен быть меньше 50 КБ, а затем отправлю его в нашу очередь обмена сообщениями. У меня есть заголовок, а затем данные.
Класс пакета:
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 recordsPartition;
private final byte replicated;
private final ByteBuffer itemBuffer = ByteBuffer.allocate(MAX_SIZE);
private int pendingItems = 0;
public Packet(final RecordPartition recordPartition) {
this.recordsPartition = (byte) recordPartition.getPartition();
this.dataCenter = Utils.LOCATION.getDatacenter();
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(recordsPartition)
.put(replicated);
}
private void sendData() {
if (itemBuffer.position() == 0) {
// no data to be sent
return;
}
final ByteBuffer buffer = ByteBuffer.allocate(MAX_SIZE);
addHeader(buffer, pendingItems);
// below line throws "BufferOverflowException"
buffer.put(itemBuffer);
SendRecord.getInstance().sendToQueueAsync(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();
}
}
}
В приведенном выше коде я получаю java.nio.BufferOverflowException в buffer.put(itemBuffer); методом sendData. Я не могу понять, почему я получаю это исключение и как его исправить.
Вот как я называю этот код:
Packet packet = new Packet(partition);
packet.addAndSendJunked("hello".getBytes(StandardCharsets.UTF_8), StringUtils.EMPTY.getBytes(StandardCharsets.UTF_8));
packet.close();
вы должны иметь возможность установить точку останова на этой строке и увидеть размеры обоих




Очевидно, что itemBuffer плюс длина заголовков, уже помещенных в buffer, превышает buffer.capacity(), то есть MAX_SIZE.
Ergo либо есть ошибка в addHeader(), либо что-то еще создает itemBuffer, либо MAX_SIZE слишком мал.
NB
SendRecord.getInstance().sendToQueueAsync(address, buffer.array());
Здесь вы ставите в очередь весь buffer размером MAX_SIZE байтов, даже если вы, возможно, не поместили в него MAX_SIZE байтов. Было бы лучше передать этому методу сам buffer и избежать бесконечных оберток в вашем коде.
Я думаю, нам нужно перевернуть «itemBuffer», прежде чем помещать его в «буфер» после его дальнейшего анализа. Также как я могу извлечь только то, что было заполнено, вместо того, чтобы отправлять все в метод sendToQueueAsync?
@ user1950349 Это правильно: flip() должен предшествовать get() или write(), или put() в другой буфер. Я уже ответил на ваш последний вопрос: не надо.
это означает, что ваш буфер имеет меньший размер, чем itemBuffer