Morph MORPH
SIGN IN SIGN UP
apache / rocketmq UNCLAIMED

Apache RocketMQ is a cloud native messaging and streaming platform, making it simple to build event-driven applications.

22393 0 0 Java
Normal View History Raw
[RIP-9] Add the introduction of the batch message samples in RocketMQ (#792) [RIP-9] Add the introduction of the batch message samples in RocketMQ
2019-02-19 17:45:03 +08:00
# Batch Message Sample
------
Sending messages in batch improves performance of delivering small messages. Messages of the same batch should have: same topic, same waitStoreMsgOK and no schedule support. You can send messages up to 4MiB at a time, but if you need to send a larger message, it is recommended to divide the larger messages into multiple small messages of no more than 1MiB.
### 1 Send Batch Messages
If you just send messages of no more than 4MiB at a time, it is easy to use batch:
```java
String topic = "BatchTest";
List<Message> messages = new ArrayList<>();
messages.add(new Message(topic, "TagA", "OrderID001", "Hello world 0".getBytes()));
messages.add(new Message(topic, "TagA", "OrderID002", "Hello world 1".getBytes()));
messages.add(new Message(topic, "TagA", "OrderID003", "Hello world 2".getBytes()));
try {
producer.send(messages);
} catch (Exception e) {
e.printStackTrace();
//handle the error
}
```
### 2 Split into Lists
The complexity only grow when you send large batch and you may not sure if it exceeds the size limit (4MiB). At this time, you’d better split the lists:
```java
[ISSUE #1505]Fix the ListSplitter of batch sample in docs
2020-03-30 17:50:24 +08:00
public class ListSplitter implements Iterator<List<Message>> {
private final int SIZE_LIMIT = 1024 * 1024 * 4;
[RIP-9] Add the introduction of the batch message samples in RocketMQ (#792) [RIP-9] Add the introduction of the batch message samples in RocketMQ
2019-02-19 17:45:03 +08:00
private final List<Message> messages;
private int currIndex;
[ISSUE #1505]Fix the ListSplitter of batch sample in docs
2020-03-30 17:50:24 +08:00
public ListSplitter(List<Message> messages) {
this.messages = messages;
[RIP-9] Add the introduction of the batch message samples in RocketMQ (#792) [RIP-9] Add the introduction of the batch message samples in RocketMQ
2019-02-19 17:45:03 +08:00
}
[ISSUE #4709] fix method is on the same line as @Override annotation (#4717) Co-authored-by: Aaron Ai <yangkun.ayk@gmail.com>
2022-08-01 20:42:34 +08:00
@Override
public boolean hasNext() {
[ISSUE #1505]Fix the ListSplitter of batch sample in docs
2020-03-30 17:50:24 +08:00
return currIndex < messages.size();
[RIP-9] Add the introduction of the batch message samples in RocketMQ (#792) [RIP-9] Add the introduction of the batch message samples in RocketMQ
2019-02-19 17:45:03 +08:00
}
[ISSUE #4709] fix method is on the same line as @Override annotation (#4717) Co-authored-by: Aaron Ai <yangkun.ayk@gmail.com>
2022-08-01 20:42:34 +08:00
@Override
public List<Message> next() {
[ISSUE #1505]Fix the ListSplitter of batch sample in docs
2020-03-30 17:50:24 +08:00
int startIndex = getStartIndex();
int nextIndex = startIndex;
[RIP-9] Add the introduction of the batch message samples in RocketMQ (#792) [RIP-9] Add the introduction of the batch message samples in RocketMQ
2019-02-19 17:45:03 +08:00
int totalSize = 0;
for (; nextIndex < messages.size(); nextIndex++) {
[ISSUE #1505]Fix the ListSplitter of batch sample in docs
2020-03-30 17:50:24 +08:00
Message message = messages.get(nextIndex);
int tmpSize = calcMessageSize(message);
[RIP-9] Add the introduction of the batch message samples in RocketMQ (#792) [RIP-9] Add the introduction of the batch message samples in RocketMQ
2019-02-19 17:45:03 +08:00
if (tmpSize + totalSize > SIZE_LIMIT) {
[ISSUE #1505]Fix the ListSplitter of batch sample in docs
2020-03-30 17:50:24 +08:00
break;
[RIP-9] Add the introduction of the batch message samples in RocketMQ (#792) [RIP-9] Add the introduction of the batch message samples in RocketMQ
2019-02-19 17:45:03 +08:00
} else {
[ISSUE #1505]Fix the ListSplitter of batch sample in docs
2020-03-30 17:50:24 +08:00
totalSize += tmpSize;
[RIP-9] Add the introduction of the batch message samples in RocketMQ (#792) [RIP-9] Add the introduction of the batch message samples in RocketMQ
2019-02-19 17:45:03 +08:00
}
}
[ISSUE #1505]Fix the ListSplitter of batch sample in docs
2020-03-30 17:50:24 +08:00
List<Message> subList = messages.subList(startIndex, nextIndex);
[RIP-9] Add the introduction of the batch message samples in RocketMQ (#792) [RIP-9] Add the introduction of the batch message samples in RocketMQ
2019-02-19 17:45:03 +08:00
currIndex = nextIndex;
return subList;
}
[ISSUE #1505]Fix the ListSplitter of batch sample in docs
2020-03-30 17:50:24 +08:00
private int getStartIndex() {
Message currMessage = messages.get(currIndex);
int tmpSize = calcMessageSize(currMessage);
while(tmpSize > SIZE_LIMIT) {
currIndex += 1;
Message message = messages.get(curIndex);
tmpSize = calcMessageSize(message);
}
return currIndex;
}
private int calcMessageSize(Message message) {
fix document mistake (#3304) message.getBody() return byte[], so change to length.
2021-09-24 17:39:54 +08:00
int tmpSize = message.getTopic().length() + message.getBody().length;
[ISSUE #1505]Fix the ListSplitter of batch sample in docs
2020-03-30 17:50:24 +08:00
Map<String, String> properties = message.getProperties();
for (Map.Entry<String, String> entry : properties.entrySet()) {
tmpSize += entry.getKey().length() + entry.getValue().length();
}
tmpSize = tmpSize + 20; // Increase the log overhead by 20 bytes
return tmpSize;
}
[RIP-9] Add the introduction of the batch message samples in RocketMQ (#792) [RIP-9] Add the introduction of the batch message samples in RocketMQ
2019-02-19 17:45:03 +08:00
}
// then you could split the large list into small ones:
ListSplitter splitter = new ListSplitter(messages);
while (splitter.hasNext()) {
try {
List<Message> listItem = splitter.next();
producer.send(listItem);
} catch (Exception e) {
e.printStackTrace();
// handle the error
}
}
fix document mistake (#3304) message.getBody() return byte[], so change to length.
2021-09-24 17:39:54 +08:00
```
Reference in New Issue Copy Permalink