407 lines
16 KiB
Java
407 lines
16 KiB
Java
package badge;
|
||
|
||
import com.red.circle.OtherServiceApplication;
|
||
import com.red.circle.framework.dto.PageQuery;
|
||
import com.red.circle.framework.dto.PageResult;
|
||
import com.red.circle.mq.business.model.event.BadgeOperateEvent;
|
||
import com.red.circle.mq.rocket.business.producer.UserMqMessageService;
|
||
import com.red.circle.other.infra.database.mongo.entity.team.team.TeamMember;
|
||
import com.red.circle.other.infra.database.mongo.service.team.team.TeamMemberService;
|
||
import com.red.circle.other.infra.database.rds.entity.badge.BadgeBackpack;
|
||
import com.red.circle.other.infra.database.rds.entity.badge.BadgeConfig;
|
||
import com.red.circle.other.infra.database.rds.entity.sys.Administrator;
|
||
import com.red.circle.other.infra.database.rds.entity.team.BusinessDevelopmentBaseInfo;
|
||
import com.red.circle.other.infra.database.rds.entity.team.RoomBdLead;
|
||
import com.red.circle.other.infra.database.rds.service.badge.BadgeBackpackService;
|
||
import com.red.circle.other.infra.database.rds.service.badge.BadgeConfigService;
|
||
import com.red.circle.other.infra.database.rds.service.sys.AdministratorService;
|
||
import com.red.circle.other.infra.database.rds.service.team.BusinessDevelopmentBaseInfoService;
|
||
import com.red.circle.other.infra.database.rds.service.team.RoomBdLeadService;
|
||
import com.red.circle.other.inner.enums.material.BadgeKeyEnum;
|
||
import com.red.circle.other.inner.enums.team.TeamMemberRole;
|
||
import com.red.circle.wallet.inner.endpoint.freight.FreightGoldClient;
|
||
import com.red.circle.wallet.inner.model.cmd.UserFreightBalancePageQryCmd;
|
||
import com.red.circle.wallet.inner.model.dto.UserFreightBalanceDTO;
|
||
import lombok.extern.slf4j.Slf4j;
|
||
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.test.context.junit4.SpringRunner;
|
||
|
||
import java.util.Collection;
|
||
import java.util.List;
|
||
import java.util.concurrent.ExecutorService;
|
||
import java.util.concurrent.Executors;
|
||
import java.util.concurrent.TimeUnit;
|
||
import java.util.concurrent.atomic.AtomicInteger;
|
||
|
||
@Slf4j
|
||
@SpringBootTest(classes = OtherServiceApplication.class)
|
||
@RunWith(SpringRunner.class)
|
||
public class BatchSendBadgeTest {
|
||
|
||
@Autowired
|
||
private TeamMemberService teamMemberService;
|
||
@Autowired
|
||
private UserMqMessageService userMqMessageService;
|
||
@Autowired
|
||
private BusinessDevelopmentBaseInfoService businessDevelopmentBaseInfoService;
|
||
@Autowired
|
||
private RoomBdLeadService roomBdLeadService;
|
||
@Autowired
|
||
private AdministratorService administratorService;
|
||
@Autowired
|
||
private FreightGoldClient freightGoldClient;
|
||
@Autowired
|
||
private BadgeBackpackService badgeBackpackService;
|
||
@Autowired
|
||
private BadgeConfigService badgeConfigService;
|
||
|
||
// 线程池大小
|
||
private static final int THREAD_POOL_SIZE = 10;
|
||
// 每批次处理间隔（毫秒）
|
||
private static final long BATCH_INTERVAL_MS = 200;
|
||
|
||
@Test
|
||
public void batchSendBadgeTest() throws InterruptedException {
|
||
ExecutorService executor = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
|
||
|
||
try {
|
||
// 1. 批量发放代理主播勋章 (HOST + AGENCY)
|
||
// sendHostAndAgencyBadges(executor);
|
||
|
||
// 2. 批量发放BD LEADER勋章
|
||
sendBdLeaderBadges(executor);
|
||
|
||
// 3. 批量发放BD勋章
|
||
// sendBdBadges(executor);
|
||
|
||
// 4. 批量发放管理员勋章
|
||
// sendAdminBadges(executor);
|
||
|
||
// 5. 批量发放金币代理勋章
|
||
// sendRechargeAgencyBadges(executor);
|
||
|
||
} finally {
|
||
executor.shutdown();
|
||
executor.awaitTermination(1, TimeUnit.HOURS);
|
||
log.info("所有勋章发放任务完成");
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 发放代理主播勋章.
|
||
*/
|
||
private void sendHostAndAgencyBadges(ExecutorService executor) {
|
||
List<TeamMember> list = teamMemberService.list();
|
||
log.info("开始发放代理主播勋章，总数: {}", list.size());
|
||
|
||
BadgeConfig agencyConfig = badgeConfigService.getByBadgeKey(BadgeKeyEnum.AGENCY.name());
|
||
if (agencyConfig == null) {
|
||
log.error("{}徽章不存在", BadgeKeyEnum.AGENCY.name());
|
||
return;
|
||
}
|
||
|
||
BadgeConfig hostConfig = badgeConfigService.getByBadgeKey(BadgeKeyEnum.HOST.name());
|
||
if (hostConfig == null) {
|
||
log.error("{}徽章不存在", BadgeKeyEnum.HOST.name());
|
||
return;
|
||
}
|
||
|
||
AtomicInteger counter = new AtomicInteger(0);
|
||
|
||
list.forEach(teamMember -> {
|
||
executor.submit(() -> {
|
||
try {
|
||
Long userId = teamMember.getMemberId();
|
||
|
||
BadgeBackpack existBadge = badgeBackpackService.query()
|
||
.eq(BadgeBackpack::getUserId, userId)
|
||
.eq(BadgeBackpack::getBadgeId, hostConfig.getId())
|
||
.getOne();
|
||
if (existBadge == null) {
|
||
// 发放主播勋章
|
||
userMqMessageService.sendBadgeOperateEvent(
|
||
userId,
|
||
BadgeKeyEnum.HOST.name(),
|
||
BadgeOperateEvent.OperationType.WEAR.name()
|
||
);
|
||
}
|
||
|
||
if (teamMember.getRole() == TeamMemberRole.OWN) {
|
||
BadgeBackpack badgeBackpack = badgeBackpackService.query()
|
||
.eq(BadgeBackpack::getUserId, userId)
|
||
.eq(BadgeBackpack::getBadgeId, agencyConfig.getId())
|
||
.getOne();
|
||
if (badgeBackpack == null) {
|
||
// 发放代理勋章
|
||
userMqMessageService.sendBadgeOperateEvent(
|
||
userId,
|
||
BadgeKeyEnum.AGENCY.name(),
|
||
BadgeOperateEvent.OperationType.WEAR.name()
|
||
);
|
||
}
|
||
}
|
||
|
||
int count = counter.incrementAndGet();
|
||
if (count % 100 == 0) {
|
||
log.info("代理主播勋章发放进度: {}/{}", count, list.size());
|
||
}
|
||
|
||
// 控制发送速率
|
||
Thread.sleep(BATCH_INTERVAL_MS);
|
||
} catch (Exception e) {
|
||
log.error("发放代理主播勋章失败, userId: {}", teamMember.getMemberId(), e);
|
||
}
|
||
});
|
||
});
|
||
|
||
log.info("代理主播勋章发放任务提交完成");
|
||
}
|
||
|
||
/**
|
||
* 发放BD Leader勋章.
|
||
*/
|
||
private void sendBdLeaderBadges(ExecutorService executor) {
|
||
List<RoomBdLead> list = roomBdLeadService.list();
|
||
log.info("开始发放BD Leader勋章，总数: {}", list.size());
|
||
|
||
BadgeConfig bdLeaderConfig = badgeConfigService.getByBadgeKey(BadgeKeyEnum.BD_LEADER.name());
|
||
if (bdLeaderConfig == null) {
|
||
log.error("{}徽章不存在", BadgeKeyEnum.BD_LEADER.name());
|
||
return;
|
||
}
|
||
|
||
|
||
AtomicInteger counter = new AtomicInteger(0);
|
||
|
||
list.forEach(bdLead -> {
|
||
executor.submit(() -> {
|
||
try {
|
||
Long userId = bdLead.getUserId();
|
||
|
||
BadgeBackpack existBadge = badgeBackpackService.query()
|
||
.eq(BadgeBackpack::getUserId, userId)
|
||
.eq(BadgeBackpack::getBadgeId, bdLeaderConfig.getId())
|
||
.getOne();
|
||
if (existBadge == null) {
|
||
// 发放BD勋章
|
||
userMqMessageService.sendBadgeOperateEvent(
|
||
userId,
|
||
BadgeKeyEnum.BD_LEADER.name(),
|
||
BadgeOperateEvent.OperationType.WEAR.name()
|
||
);
|
||
}
|
||
|
||
int count = counter.incrementAndGet();
|
||
if (count % 100 == 0) {
|
||
log.info("BD Leader勋章发放进度: {}/{}", count, list.size());
|
||
}
|
||
|
||
Thread.sleep(BATCH_INTERVAL_MS);
|
||
} catch (Exception e) {
|
||
log.error("发放BD Leader勋章失败, userId: {}", bdLead.getUserId(), e);
|
||
}
|
||
});
|
||
});
|
||
|
||
log.info("BD Leader勋章发放任务提交完成");
|
||
}
|
||
|
||
/**
|
||
* 发放BD勋章.
|
||
*/
|
||
private void sendBdBadges(ExecutorService executor) {
|
||
List<BusinessDevelopmentBaseInfo> list = businessDevelopmentBaseInfoService.listAll();
|
||
log.info("开始发放BD勋章，总数: {}", list.size());
|
||
|
||
BadgeConfig bdConfig = badgeConfigService.getByBadgeKey(BadgeKeyEnum.BD.name());
|
||
if (bdConfig == null) {
|
||
log.error("{}徽章不存在", BadgeKeyEnum.BD.name());
|
||
return;
|
||
}
|
||
|
||
AtomicInteger counter = new AtomicInteger(0);
|
||
|
||
list.forEach(bd -> {
|
||
executor.submit(() -> {
|
||
try {
|
||
Long userId = bd.getUserId();
|
||
|
||
BadgeBackpack existBadge = badgeBackpackService.query()
|
||
.eq(BadgeBackpack::getUserId, userId)
|
||
.eq(BadgeBackpack::getBadgeId, bdConfig.getId())
|
||
.getOne();
|
||
if (existBadge == null) {
|
||
// 发放BD勋章
|
||
userMqMessageService.sendBadgeOperateEvent(
|
||
userId,
|
||
BadgeKeyEnum.BD.name(),
|
||
BadgeOperateEvent.OperationType.WEAR.name()
|
||
);
|
||
}
|
||
|
||
int count = counter.incrementAndGet();
|
||
if (count % 100 == 0) {
|
||
log.info("BD勋章发放进度: {}/{}", count, list.size());
|
||
}
|
||
|
||
Thread.sleep(BATCH_INTERVAL_MS);
|
||
} catch (Exception e) {
|
||
log.error("发放BD勋章失败, userId: {}", bd.getUserId(), e);
|
||
}
|
||
});
|
||
});
|
||
|
||
log.info("BD勋章发放任务提交完成");
|
||
}
|
||
|
||
/**
|
||
* 发放管理员勋章.
|
||
*/
|
||
private void sendAdminBadges(ExecutorService executor) {
|
||
List<Administrator> list = administratorService.list();
|
||
log.info("开始发放管理员勋章，总数: {}", list.size());
|
||
|
||
AtomicInteger counter = new AtomicInteger(0);
|
||
|
||
list.forEach(admin -> {
|
||
executor.submit(() -> {
|
||
try {
|
||
Long userId = admin.getUserId();
|
||
String roles = admin.getRoles();
|
||
if (!Boolean.TRUE.equals(admin.getStatus())) {
|
||
return;
|
||
}
|
||
|
||
if ("ADMIN".equals(roles)) {
|
||
// ADMIN角色：只发ADMIN勋章
|
||
userMqMessageService.sendBadgeOperateEvent(
|
||
userId,
|
||
BadgeKeyEnum.ADMIN.name(),
|
||
BadgeOperateEvent.OperationType.WEAR.name()
|
||
);
|
||
} else if ("SUPER_ADMIN".equals(roles)) {
|
||
// SUPER_ADMIN角色：发SUPER_ADMIN勋章
|
||
userMqMessageService.sendBadgeOperateEvent(
|
||
userId,
|
||
BadgeKeyEnum.SUPER_ADMIN.name(),
|
||
BadgeOperateEvent.OperationType.WEAR.name()
|
||
);
|
||
} else if ("MANAGER".equals(roles)) {
|
||
// MANAGER角色：发SUPER_ADMIN和ADMIN勋章
|
||
userMqMessageService.sendBadgeOperateEvent(
|
||
userId,
|
||
BadgeKeyEnum.SUPER_ADMIN.name(),
|
||
BadgeOperateEvent.OperationType.WEAR.name()
|
||
);
|
||
userMqMessageService.sendBadgeOperateEvent(
|
||
userId,
|
||
BadgeKeyEnum.ADMIN.name(),
|
||
BadgeOperateEvent.OperationType.WEAR.name()
|
||
);
|
||
}
|
||
|
||
int count = counter.incrementAndGet();
|
||
if (count % 100 == 0) {
|
||
log.info("管理员勋章发放进度: {}/{}", count, list.size());
|
||
}
|
||
|
||
Thread.sleep(BATCH_INTERVAL_MS);
|
||
} catch (Exception e) {
|
||
log.error("发放管理员勋章失败, userId: {}", admin.getUserId(), e);
|
||
}
|
||
});
|
||
});
|
||
|
||
log.info("管理员勋章发放任务提交完成");
|
||
}
|
||
|
||
/**
|
||
* 发放充值代理勋章.
|
||
*/
|
||
private void sendRechargeAgencyBadges(ExecutorService executor) {
|
||
log.info("开始发放充值代理勋章");
|
||
|
||
AtomicInteger totalCounter = new AtomicInteger(0);
|
||
int cursor = 1;
|
||
boolean hasMore = true;
|
||
|
||
BadgeConfig badgeConfig = badgeConfigService.getByBadgeKey(BadgeKeyEnum.RECHARGE_AGENCY.name());
|
||
if (badgeConfig == null) {
|
||
log.error("{}徽章不存在", BadgeKeyEnum.RECHARGE_AGENCY.name());
|
||
return;
|
||
}
|
||
|
||
while (hasMore) {
|
||
PageQuery pageQuery = new PageQuery();
|
||
pageQuery.setCursor(cursor);
|
||
pageQuery.setLimit(500);
|
||
pageQuery.setSearchCount(true);
|
||
|
||
UserFreightBalancePageQryCmd cmd = new UserFreightBalancePageQryCmd();
|
||
cmd.setClose(false);
|
||
cmd.setPageQuery(pageQuery);
|
||
|
||
try {
|
||
PageResult<UserFreightBalanceDTO> freightPageResult =
|
||
freightGoldClient.pageFreight(cmd).getBody();
|
||
Collection<UserFreightBalanceDTO> records = freightPageResult.getRecords();
|
||
|
||
if (records == null || records.isEmpty()) {
|
||
hasMore = false;
|
||
break;
|
||
}
|
||
|
||
log.info("充值代理勋章发放，当前页: {}, 本页数量: {}", cursor, records.size());
|
||
|
||
records.forEach(freight -> {
|
||
executor.submit(() -> {
|
||
try {
|
||
Long userId = freight.getUserId();
|
||
if (freight.getClose()) {
|
||
return;
|
||
}
|
||
|
||
BadgeBackpack existBadge = badgeBackpackService.query()
|
||
.eq(BadgeBackpack::getUserId, userId)
|
||
.eq(BadgeBackpack::getBadgeId, badgeConfig.getId())
|
||
.getOne();
|
||
if (existBadge != null) {
|
||
return;
|
||
}
|
||
|
||
userMqMessageService.sendBadgeOperateEvent(
|
||
userId,
|
||
BadgeKeyEnum.RECHARGE_AGENCY.name(),
|
||
BadgeOperateEvent.OperationType.WEAR.name()
|
||
);
|
||
|
||
int count = totalCounter.incrementAndGet();
|
||
if (count % 100 == 0) {
|
||
log.info("充值代理勋章发放进度: {}", count);
|
||
}
|
||
|
||
Thread.sleep(BATCH_INTERVAL_MS);
|
||
} catch (Exception e) {
|
||
log.error("发放充值代理勋章失败, userId: {}", freight.getUserId(), e);
|
||
}
|
||
});
|
||
});
|
||
|
||
cursor++;
|
||
|
||
// 检查是否还有下一页
|
||
hasMore = freightPageResult.getTotal() > cursor * 500;
|
||
|
||
} catch (Exception e) {
|
||
log.error("查询充值代理列表失败, cursor: {}", cursor, e);
|
||
hasMore = false;
|
||
}
|
||
}
|
||
|
||
log.info("充值代理勋章发放任务提交完成，总数: {}", totalCounter.get());
|
||
}
|
||
}
|