gtsoft/gtsoft-snail-job-server
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects 1 Releases Wiki Activity

feat:2.3.0

Browse Source 
1. 修复当一个组由打开到关闭，已分配的组未能清除问题
2. 修复迁移死信队列id重复问题
This commit is contained in:
byteblogs168 2023-09-06 22:43:32 +08:00
parent 8121126cdf
commit 408a0f2699
4 changed files with 18 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/service/convert/RetryDeadLetterConverter.java
View File

@ -2,9 +2,7 @@ package com.aizuda.easy.retry.server.service.convert;
import com.aizuda.easy.retry.template.datasource.persistence.po.RetryDeadLetter;
import com.aizuda.easy.retry.template.datasource.persistence.po.RetryTask;
import org.mapstruct.Mapper;
import org.mapstruct.Mapping;
import org.mapstruct.Mappings;
import org.mapstruct.*;
import org.mapstruct.factory.Mappers;
import java.util.List;
@ -19,10 +17,14 @@ public interface RetryDeadLetterConverter {
RetryDeadLetterConverter INSTANCE = Mappers.getMapper(RetryDeadLetterConverter.class);
@Named("ignoreId")
@Mappings({
@Mapping(source = "id", target = "id", ignore = true),
@Mapping(source = "createDt", target = "createDt", ignore = true)
@Mapping(target = "id", ignore = true),
@Mapping(target = "createDt", ignore = true)
})
RetryDeadLetter toRetryDeadLetter(RetryTask retryTasks);
@IterableMapping(qualifiedByName = "ignoreId")
List<RetryDeadLetter> toRetryDeadLetter(List<RetryTask> retryTasks);
}

5
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/service/impl/RetryServiceImpl.java
View File

@ -237,7 +237,10 @@ public class RetryServiceImpl implements RetryService {
}
List<RetryDeadLetter> retryDeadLetters = RetryDeadLetterConverter.INSTANCE.toRetryDeadLetter(retryTasks);
LocalDateTime now = LocalDateTime.now();
for (RetryDeadLetter retryDeadLetter : retryDeadLetters) {
retryDeadLetter.setCreateDt(now);
}
Assert.isTrue(retryDeadLetters.size() == accessTemplate
.getRetryDeadLetterAccess().batchInsert(groupName, retryDeadLetters),
() -> new EasyRetryServerException("插入死信队列失败 [{}]", JsonUtil.toJsonString(retryDeadLetters)));

4
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/support/allocate/server/AllocateMessageQueueConsistentHash.java
View File

@ -58,10 +58,10 @@ public class AllocateMessageQueueConsistentHash implements ServerLoadBalance {
throw new IllegalArgumentException("currentCID is empty");
}
if (CollectionUtils.isEmpty(groupList)) {
throw new IllegalArgumentException("mqAll is null or mqAll empty");
throw new IllegalArgumentException("groupList is null or groupList empty");
}
if (CollectionUtils.isEmpty(serverList)) {
throw new IllegalArgumentException("cidAll is null or cidAll empty");
throw new IllegalArgumentException("serverList is null or serverList empty");
}
List<String> result = new ArrayList<>();

8
easy-retry-server/src/main/java/com/aizuda/easy/retry/server/support/handler/ServerNodeBalance.java
View File

@ -72,20 +72,22 @@ public class ServerNodeBalance implements Lifecycle, Runnable {
if (CollectionUtils.isEmpty(podIpSet)) {
LogUtils.error(log, "server node is empty");
return;
}
Set<String> allGroup = CacheGroup.getAllGroup();
if (CollectionUtils.isEmpty(allGroup)) {
LogUtils.error(log, "group is empty");
}
// 删除本地缓存的所有组信息
CacheConsumerGroup.clear();
if(CollectionUtils.isEmpty(podIpSet) || CollectionUtils.isEmpty(allGroup)) {
return;
}
List<String> allocate = new AllocateMessageQueueConsistentHash()
.allocate(ServerRegister.CURRENT_CID, new ArrayList<>(allGroup), new ArrayList<>(podIpSet));
// 删除本地缓存的所有组信息
CacheConsumerGroup.clear();
// 重新覆盖本地分配的组信息
for (String groupName : allocate) {
CacheConsumerGroup.addOrUpdate(groupName);