wangli/im-center
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

REQ-2481: 添加删除日志表的job

Browse Source
This commit is contained in:
yanglin 2024-05-31 15:08:25 +08:00
parent 815960d79b
commit 9c0a99c18e
1 changed files with 20 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
im-center-server/src/main/java/cn/axzo/im/send/SendQueue.java
View File

@ -70,25 +70,12 @@ public class SendQueue {
return histories;
}
// 批量最大的优先, 不要浪费批量接口的流量
Map<String, List<Record>> batchNo2Records = new HashMap<>();
boolean hasBatchRecords = false;
for (Record record : records) {
String batchNo = record.history.determineBatchNo().orElse(null);
if (batchNo == null) continue;
List<Record> batchRecords = batchNo2Records
.computeIfAbsent(batchNo, unused -> new ArrayList<>());
batchRecords.add(record);
hasBatchRecords = hasBatchRecords || batchRecords.size() > 1;
List<Record> perfectRecords = getQueuedMaxBatch();
// 如果没有真正的批量, 就退回到按优先级顺序发送
if (perfectRecords.size() <= 1) {
MessageHistory history = poll().orElse(null);
return Collections.singletonList(history);
}
if (!hasBatchRecords) {
// as poll one
MessageHistory record = poll().orElse(null);
return Collections.singletonList(record);
}
List<List<Record>> groups = new ArrayList<>(batchNo2Records.values());
groups.sort(Comparator.comparingInt(List::size));
// a batch that has max records
List<Record> perfectRecords = groups.get(groups.size() - 1);
int minSize = Math.min(batchSize, perfectRecords.size());
ArrayList<Record> pollRecords = new ArrayList<>(minSize);
for (int i = 0; i < minSize; i++) {
@ -117,6 +104,21 @@ public class SendQueue {
return histories;
}
private List<Record> getQueuedMaxBatch() {
Map<String, List<Record>> batchNo2Records = new HashMap<>();
for (Record record : records) {
String batchNo = record.history.determineBatchNo().orElse(null);
if (batchNo == null) continue;
List<Record> batchRecords = batchNo2Records
.computeIfAbsent(batchNo, unused -> new ArrayList<>());
batchRecords.add(record);
}
List<List<Record>> groups = new ArrayList<>(batchNo2Records.values());
groups.sort(Comparator.comparingInt(List::size));
// a batch that has max records
return groups.get(groups.size() - 1);
}
public synchronized Optional<MessageHistory> poll() {
if (determineNoMoreRecords())
return Optional.empty();