未验证 提交 92b8cf97 编写于 作者: M MountainOne 提交者: GitHub

[ISSUE #1505]Fix the ListSplitter of batch sample in docs

上级 30ad1723
...@@ -500,7 +500,6 @@ try { ...@@ -500,7 +500,6 @@ try {
复杂度只有当你发送大批量时才会增长,你可能不确定它是否超过了大小限制(4MB)。这时候你最好把你的消息列表分割一下: 复杂度只有当你发送大批量时才会增长,你可能不确定它是否超过了大小限制(4MB)。这时候你最好把你的消息列表分割一下:
```java ```java
public class ListSplitter implements Iterator<List<Message>> { public class ListSplitter implements Iterator<List<Message>> {
private final int SIZE_LIMIT = 1024 * 1024 * 4; private final int SIZE_LIMIT = 1024 * 1024 * 4;
private final List<Message> messages; private final List<Message> messages;
...@@ -512,36 +511,41 @@ public class ListSplitter implements Iterator<List<Message>> { ...@@ -512,36 +511,41 @@ public class ListSplitter implements Iterator<List<Message>> {
return currIndex < messages.size(); return currIndex < messages.size();
} }
@Override public List<Message> next() { @Override public List<Message> next() {
int nextIndex = currIndex; int startIndex = getStartIndex();
int nextIndex = startIndex;
int totalSize = 0; int totalSize = 0;
for (; nextIndex < messages.size(); nextIndex++) { for (; nextIndex < messages.size(); nextIndex++) {
Message message = messages.get(nextIndex); Message message = messages.get(nextIndex);
int tmpSize = message.getTopic().length() + message.getBody().length; int tmpSize = calcMessageSize(message);
Map<String, String> properties = message.getProperties();
for (Map.Entry<String, String> entry : properties.entrySet()) {
tmpSize += entry.getKey().length() + entry.getValue().length();
}
tmpSize = tmpSize + 20; // 增加日志的开销20字节
if (tmpSize > SIZE_LIMIT) {
//单个消息超过了最大的限制
//忽略,否则会阻塞分裂的进程
if (nextIndex - currIndex == 0) {
//假如下一个子列表没有元素,则添加这个子列表然后退出循环,否则只是退出循环
nextIndex++;
}
break;
}
if (tmpSize + totalSize > SIZE_LIMIT) { if (tmpSize + totalSize > SIZE_LIMIT) {
break; break;
} else { } else {
totalSize += tmpSize; totalSize += tmpSize;
} }
} }
List<Message> subList = messages.subList(currIndex, nextIndex); List<Message> subList = messages.subList(startIndex, nextIndex);
currIndex = nextIndex; currIndex = nextIndex;
return subList; return subList;
} }
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) {
int tmpSize = message.getTopic().length() + message.getBody().length();
Map<String, String> properties = message.getProperties();
for (Map.Entry<String, String> entry : properties.entrySet()) {
tmpSize += entry.getKey().length() + entry.getValue().length();
}
tmpSize = tmpSize + 20; // 增加⽇日志的开销20字节
return tmpSize;
}
} }
//把大的消息分裂成若干个小的消息 //把大的消息分裂成若干个小的消息
ListSplitter splitter = new ListSplitter(messages); ListSplitter splitter = new ListSplitter(messages);
......
...@@ -20,7 +20,7 @@ try { ...@@ -20,7 +20,7 @@ try {
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: 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 ```java
public class ListSplitter implements Iterator<List<Message>> { public class ListSplitter implements Iterator<List<Message>> {
private final int SIZE_LIMIT = 1000 * 1000; private final int SIZE_LIMIT = 1024 * 1024 * 4;
private final List<Message> messages; private final List<Message> messages;
private int currIndex; private int currIndex;
public ListSplitter(List<Message> messages) { public ListSplitter(List<Message> messages) {
...@@ -30,36 +30,41 @@ public class ListSplitter implements Iterator<List<Message>> { ...@@ -30,36 +30,41 @@ public class ListSplitter implements Iterator<List<Message>> {
return currIndex < messages.size(); return currIndex < messages.size();
} }
@Override public List<Message> next() { @Override public List<Message> next() {
int nextIndex = currIndex; int startIndex = getStartIndex();
int nextIndex = startIndex;
int totalSize = 0; int totalSize = 0;
for (; nextIndex < messages.size(); nextIndex++) { for (; nextIndex < messages.size(); nextIndex++) {
Message message = messages.get(nextIndex); Message message = messages.get(nextIndex);
int tmpSize = message.getTopic().length() + message.getBody().length; int tmpSize = calcMessageSize(message);
Map<String, String> properties = message.getProperties();
for (Map.Entry<String, String> entry : properties.entrySet()) {
tmpSize += entry.getKey().length() + entry.getValue().length();
}
tmpSize = tmpSize + 20; //for log overhead
if (tmpSize > SIZE_LIMIT) {
//it is unexpected that single message exceeds the SIZE_LIMIT
//here just let it go, otherwise it will block the splitting process
if (nextIndex - currIndex == 0) {
//if the next sublist has no element, add this one and then break, otherwise just break
nextIndex++;
}
break;
}
if (tmpSize + totalSize > SIZE_LIMIT) { if (tmpSize + totalSize > SIZE_LIMIT) {
break; break;
} else { } else {
totalSize += tmpSize; totalSize += tmpSize;
} }
} }
List<Message> subList = messages.subList(currIndex, nextIndex); List<Message> subList = messages.subList(startIndex, nextIndex);
currIndex = nextIndex; currIndex = nextIndex;
return subList; return subList;
} }
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) {
int tmpSize = message.getTopic().length() + message.getBody().length();
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;
}
} }
// then you could split the large list into small ones: // then you could split the large list into small ones:
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册