mq消息重试功能调试
This commit is contained in:
tianfeng
parent
2a05a4ad00
commit
7c0f44f8a5
@ -11,6 +11,7 @@ import com.red.circle.tool.core.text.StringUtils;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.Objects;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
/**
|
||||
@ -27,6 +28,7 @@ public class MessageSenderService {
|
||||
|
||||
public MessageSenderService(
|
||||
ManagedRocketMQProducer managedRocketMQProducer,
|
||||
@Qualifier("rocketMqSenderFailServiceRedisImpl")
|
||||
RocketMqSenderFailService rocketMqSenderFailService) {
|
||||
this.managedRocketMQProducer = managedRocketMQProducer;
|
||||
this.rocketMqSenderFailService = rocketMqSenderFailService;
|
||||
|
||||
@ -38,7 +38,7 @@ public class GiftMqRetryTask {
|
||||
* 每分钟重试赠送礼物消息
|
||||
*/
|
||||
@Scheduled(cron = "0 */1 * * * ?")
|
||||
@TaskCacheLock(key = "GIFT_MQ_RETRY_TASK_GIVE_GIFT", expireSecond = 60)
|
||||
@TaskCacheLock(key = "GIFT_MQ_RETRY_TASK_GIVE_GIFT", expireSecond = 55)
|
||||
public void retryGiveGiftMessages() {
|
||||
try {
|
||||
List<RocketMqFailMessageDetail> messages = queryService.getPendingMessagesByTag(
|
||||
@ -63,7 +63,7 @@ public class GiftMqRetryTask {
|
||||
* 每分钟重试离线统计礼物消息
|
||||
*/
|
||||
@Scheduled(cron = "0 */1 * * * ?")
|
||||
@TaskCacheLock(key = "GIFT_MQ_RETRY_TASK_OFFLINE_PROCESS", expireSecond = 60)
|
||||
@TaskCacheLock(key = "GIFT_MQ_RETRY_TASK_OFFLINE_PROCESS", expireSecond = 55)
|
||||
public void retryOfflineProcessGiftMessages() {
|
||||
try {
|
||||
List<RocketMqFailMessageDetail> messages = queryService.getPendingMessagesByTag(
|
||||
@ -142,13 +142,17 @@ public class GiftMqRetryTask {
|
||||
}
|
||||
}
|
||||
|
||||
private GiveAwayGiftBatchEvent parseGiveGiftEvent(
|
||||
String body) {
|
||||
private GiveAwayGiftBatchEvent parseGiveGiftEvent(String body) {
|
||||
if (body.startsWith("\"") && body.endsWith("\"")) {
|
||||
body = JSON.parseObject(body, String.class);
|
||||
}
|
||||
return JSON.parseObject(body, GiveAwayGiftBatchEvent.class);
|
||||
}
|
||||
|
||||
private OfflineProcessGiftEvent parseOfflineProcessEvent(
|
||||
String body) {
|
||||
private OfflineProcessGiftEvent parseOfflineProcessEvent(String body) {
|
||||
if (body.startsWith("\"") && body.endsWith("\"")) {
|
||||
body = JSON.parseObject(body, String.class);
|
||||
}
|
||||
return JSON.parseObject(body, OfflineProcessGiftEvent.class);
|
||||
}
|
||||
}
|
||||
|
||||
@ -0,0 +1,131 @@
|
||||
import com.red.circle.OtherServiceApplication;
|
||||
import com.red.circle.other.infra.common.team.decay.ListMemberWorkDecay;
|
||||
import com.red.circle.other.infra.database.mongo.entity.team.team.TeamMemberTarget;
|
||||
import com.red.circle.other.infra.database.mongo.service.team.team.TeamMemberTargetService;
|
||||
import com.red.circle.other.inner.model.cmd.team.TeamMemberWorkQryCmd;
|
||||
import com.red.circle.other.inner.model.dto.agency.agency.TeamMemberWorkDTO;
|
||||
import com.red.circle.other.inner.model.dto.agency.agency.TeamMemberWorkDailyTargetDTO;
|
||||
import com.red.circle.tool.core.date.TimestampUtils;
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.data.mongodb.core.MongoTemplate;
|
||||
import org.springframework.data.mongodb.core.query.Criteria;
|
||||
import org.springframework.data.mongodb.core.query.Query;
|
||||
import org.springframework.data.mongodb.core.query.Update;
|
||||
import org.springframework.test.context.junit4.SpringRunner;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
@SpringBootTest(classes = OtherServiceApplication.class)
|
||||
@RunWith(SpringRunner.class)
|
||||
public class AnchorTargetTest {
|
||||
|
||||
@Autowired
|
||||
private TeamMemberTargetService teamMemberTargetService;
|
||||
@Autowired
|
||||
private ListMemberWorkDecay listMemberWorkDecay;
|
||||
@Autowired
|
||||
private MongoTemplate mongoTemplate;
|
||||
|
||||
|
||||
/**
|
||||
* 扣除主播目标
|
||||
*/
|
||||
@Test
|
||||
public void AnchorTargetTest(){
|
||||
Long userId = 1987522531364773890L;
|
||||
Long deductValue = 300000L; // 要扣除的目标值
|
||||
|
||||
TeamMemberWorkQryCmd cmd = new TeamMemberWorkQryCmd();
|
||||
cmd.setMemberUserId(userId);
|
||||
cmd.setSysOrigin("LIKEI");
|
||||
List<TeamMemberWorkDTO> execute = listMemberWorkDecay.execute(cmd);
|
||||
|
||||
// 1. 筛选 billStatus = UNPAID 的数据
|
||||
List<TeamMemberWorkDTO> unpaidList = execute.stream()
|
||||
.filter(work -> "UNPAID".equals(work.getBillStatus()))
|
||||
.limit(1)
|
||||
.toList();
|
||||
|
||||
System.out.println("未结算的工单数量: " + unpaidList.size());
|
||||
|
||||
for (TeamMemberWorkDTO work : unpaidList) {
|
||||
// 2. 统计 dailyTargets 的 acceptGiftValue 的和
|
||||
Long totalAcceptGiftValue = work.getTarget().getDailyTargets().stream()
|
||||
.mapToLong(TeamMemberWorkDailyTargetDTO::getAcceptGiftValue)
|
||||
.sum();
|
||||
|
||||
System.out.println("工单ID: " + work.getId() + ", 总收礼物价值: " + totalAcceptGiftValue);
|
||||
|
||||
// 3. 检查目标值够不够扣
|
||||
if (totalAcceptGiftValue < deductValue) {
|
||||
System.out.println("目标值不足，无法扣除。需要: " + deductValue + ", 当前: " + totalAcceptGiftValue);
|
||||
continue;
|
||||
}
|
||||
|
||||
// 4. 从第一天开始扣除
|
||||
Long remainingDeduct = deductValue;
|
||||
List<TeamMemberWorkDailyTargetDTO> dailyTargets = work.getTarget().getDailyTargets();
|
||||
boolean hasModified = false; // 标记是否有修改
|
||||
|
||||
for (int i = 0; i < dailyTargets.size(); i++) {
|
||||
TeamMemberWorkDailyTargetDTO dailyTarget = dailyTargets.get(i);
|
||||
Long currentValue = dailyTarget.getAcceptGiftValue();
|
||||
|
||||
if (remainingDeduct <= 0) {
|
||||
break;
|
||||
}
|
||||
|
||||
if (currentValue > 0) { // 只有当前值大于0时才可能修改
|
||||
hasModified = true; // 标记为已修改
|
||||
|
||||
if (currentValue >= remainingDeduct) {
|
||||
// 当前天的值足够扣除剩余部分
|
||||
dailyTarget.setAcceptGiftValue(currentValue - remainingDeduct);
|
||||
System.out.println("日期 " + dailyTarget.getDateNumber() + " 扣除: " + remainingDeduct + ", 剩余: " + (currentValue - remainingDeduct));
|
||||
remainingDeduct = 0L;
|
||||
} else {
|
||||
// 当前天的值不够，全部扣除
|
||||
dailyTarget.setAcceptGiftValue(0L);
|
||||
remainingDeduct -= currentValue;
|
||||
System.out.println("日期 " + dailyTarget.getDateNumber() + " 扣除: " + currentValue + ", 剩余需扣除: " + remainingDeduct);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 5. 只有当有修改时才更新 MongoDB
|
||||
if (hasModified) {
|
||||
updateTeamMemberTarget(work.getId(), dailyTargets);
|
||||
System.out.println("工单 " + work.getId() + " 扣除完成，已更新到MongoDB");
|
||||
} else {
|
||||
System.out.println("工单 " + work.getId() + " 无需修改，跳过更新");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 更新 MongoDB 中的 TeamMemberTarget
|
||||
* @param timeId 工单ID
|
||||
* @param dailyTargets 更新后的每日目标列表
|
||||
*/
|
||||
private void updateTeamMemberTarget(Long timeId, List<TeamMemberWorkDailyTargetDTO> dailyTargets) {
|
||||
Query query = new Query(Criteria.where("timeId").is(timeId));
|
||||
|
||||
// 更新 target.dailyTargets 数组
|
||||
for (int i = 0; i < dailyTargets.size(); i++) {
|
||||
TeamMemberWorkDailyTargetDTO dailyTarget = dailyTargets.get(i);
|
||||
|
||||
mongoTemplate.updateFirst(query,
|
||||
new Update()
|
||||
.set("updateTime", TimestampUtils.now())
|
||||
.set("dailyTargets.$[i].acceptGiftValue", dailyTarget.getAcceptGiftValue())
|
||||
.set("dailyTargets.$[i].updateTime", TimestampUtils.now())
|
||||
.filterArray(Criteria.where("i.dateNumber").is(dailyTarget.getDateNumber())),
|
||||
TeamMemberTarget.class);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user