当前位置：首页 > news >正文

高效异步处理：基于RocketMQ的消费系统架构全解析

news 2026/3/26 17:25:51

基于 RocketMQ 的考勤统计批量处理系统核心实现

一、消息实体定义

二、生产者服务实现

三、消费者监听器实现

四、线程池配置

五、业务服务实现

六、配置文件

七、Maven依赖

八、关键设计要点

8.1 生产者发送策略

8.2 消费者处理策略

8.3 配置优化

一、消息实体定义

package jnpf.model.attendance.event; import lombok.AllArgsConstructor; import lombok.Builder; import lombok.Data; import lombok.NoArgsConstructor; import javax.validation.constraints.NotBlank; import javax.validation.constraints.NotEmpty; import javax.validation.constraints.NotNull; import java.util.Date; import java.util.List; /** * 考勤统计批量处理数据传输对象 */ @Data @Builder @AllArgsConstructor @NoArgsConstructor public class AttendanceStatisticsBatchDto { /** * 租户ID */ @NotBlank(message = "租户Id不能为空") private String tenantId; /** * 考勤组ID */ @NotBlank(message = "考勤组Id不能为空") private String groupId; /** * 用户ID集合 */ @NotEmpty(message = "用户Id集合不能为空") private List<String> userIdList; /** * 统计日期 */ @NotNull(message = "日期不能为空") private Date day; }

二、生产者服务实现

package jnpf.attendance.service.impl; import jnpf.constants.MessageTopicConstants; import jnpf.model.attendance.event.AttendanceStatisticsBatchDto; import lombok.extern.slf4j.Slf4j; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.spring.core.RocketMQTemplate; import org.springframework.messaging.Message; import org.springframework.messaging.support.MessageBuilder; import org.springframework.stereotype.Service; import javax.annotation.Resource; import com.alibaba.fastjson.JSON; /** * 考勤统计消息生产者服务实现 */ @Slf4j @Service public class AttendanceStatisticsProducerServiceImpl { @Resource private RocketMQTemplate rocketMqTemplate; /** * 同步发送批量统计消息 * * @param dto 统计参数 * @return 发送结果 */ public SendResult sendBatchStatisticsMessage(AttendanceStatisticsBatchDto dto) { String messageStr = JSON.toJSONString(dto); Message<String> message = MessageBuilder.withPayload(messageStr) .setHeader("tenantId", dto.getTenantId()) .setHeader("msgType", "ATTENDANCE_STATISTICS_BATCH") .setHeader("sendTime", System.currentTimeMillis()) .build(); log.info("发送考勤统计批量消息: tenantId={}, groupId={}, 用户数={}, 日期={}", dto.getTenantId(), dto.getGroupId(), dto.getUserIdList().size(), dto.getDay()); return rocketMqTemplate.syncSend(MessageTopicConstants.ATTENDANCE_STATISTICS_BATCH_TOPIC, message); } /** * 异步发送批量统计消息 * * @param dto 统计参数 */ public void sendBatchStatisticsMessageAsync(AttendanceStatisticsBatchDto dto) { String messageStr = JSON.toJSONString(dto); Message<String> message = MessageBuilder.withPayload(messageStr) .setHeader("tenantId", dto.getTenantId()) .setHeader("msgType", "ATTENDANCE_STATISTICS_BATCH_ASYNC") .setHeader("sendTime", System.currentTimeMillis()) .build(); log.info("异步发送考勤统计批量消息: tenantId={}, 用户数={}", dto.getTenantId(), dto.getUserIdList().size()); rocketMqTemplate.asyncSend(MessageTopicConstants.ATTENDANCE_STATISTICS_BATCH_TOPIC, message, new org.apache.rocketmq.client.producer.SendCallback() { @Override public void onSuccess(SendResult sendResult) { log.info("异步消息发送成功: msgId={}", sendResult.getMsgId()); } @Override public void onException(Throwable throwable) { log.error("异步消息发送失败: tenantId={}, error={}", dto.getTenantId(), throwable.getMessage()); } }); } /** * 发送单向消息（不关心发送结果） * * @param dto 统计参数 */ public void sendBatchStatisticsMessageOneWay(AttendanceStatisticsBatchDto dto) { String messageStr = JSON.toJSONString(dto); Message<String> message = MessageBuilder.withPayload(messageStr) .setHeader("tenantId", dto.getTenantId()) .setHeader("msgType", "ATTENDANCE_STATISTICS_BATCH_ONEWAY") .setHeader("sendTime", System.currentTimeMillis()) .build(); log.debug("单向发送考勤统计批量消息: tenantId={}, 用户数={}", dto.getTenantId(), dto.getUserIdList().size()); rocketMqTemplate.sendOneWay(MessageTopicConstants.ATTENDANCE_STATISTICS_BATCH_TOPIC, message); } /** * 发送带延迟的批量统计消息 * * @param dto 统计参数 * @param delayLevel 延迟级别 * @return 发送结果 */ public SendResult sendBatchStatisticsMessageWithDelay(AttendanceStatisticsBatchDto dto, int delayLevel) { String messageStr = JSON.toJSONString(dto); Message<String> message = MessageBuilder.withPayload(messageStr) .setHeader("tenantId", dto.getTenantId()) .setHeader("msgType", "ATTENDANCE_STATISTICS_BATCH_DELAY") .setHeader("sendTime", System.currentTimeMillis()) .build(); log.info("发送延迟考勤统计批量消息: tenantId={}, delayLevel={}, 用户数={}", dto.getTenantId(), delayLevel, dto.getUserIdList().size()); return rocketMqTemplate.syncSend(MessageTopicConstants.ATTENDANCE_STATISTICS_BATCH_TOPIC, message, 3000, delayLevel); } }

三、消费者监听器实现

package jnpf.attendance.event; import cn.hutool.core.collection.CollUtil; import cn.hutool.json.JSONUtil; import jnpf.attendance.service.AttendanceDayStatisticsService; import jnpf.constants.MessageTopicConstants; import jnpf.model.attendance.event.AttendanceStatisticsBatchDto; import jnpf.util.StringUtil; import lombok.extern.slf4j.Slf4j; import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; import org.apache.rocketmq.spring.annotation.ConsumeMode; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQListener; import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener; import org.springframework.stereotype.Component; import javax.annotation.Resource; import java.util.Objects; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; /** * 考勤统计批量消息消费者 */ @Slf4j @Component @RocketMQMessageListener( topic = MessageTopicConstants.ATTENDANCE_STATISTICS_BATCH_TOPIC, consumerGroup = MessageTopicConstants.ATTENDANCE_STATISTICS_BATCH_CONSUMER_GROUP, consumeMode = ConsumeMode.CONCURRENTLY, consumeThreadNumber = 2, replyTimeout = 600000, maxReconsumeTimes = 1) public class StatisticsBatchMQListener implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener { @Resource(name = "ioIntensiveThreadPool") private ThreadPoolExecutor threadPoolExecutor; @Resource private AttendanceDayStatisticsService attendanceDayStatisticsService; @Override public void onMessage(String message) { log.info("接收到批量生成用户日统计消息，参数为:{}", message); // 1. 空消息检查 if (StringUtil.isEmpty(message)) { log.warn("接收到空消息，跳过处理"); return; } // 2. JSON反序列化 AttendanceStatisticsBatchDto batchDto; try { batchDto = JSONUtil.toBean(message, AttendanceStatisticsBatchDto.class); } catch (Exception e) { log.error("消息反序列化失败，消息内容: {}", message, e); return; } // 3. 参数完整性验证 if (Objects.isNull(batchDto) || CollUtil.isEmpty(batchDto.getUserIdList()) || StringUtil.isEmpty(batchDto.getTenantId()) || StringUtil.isEmpty(batchDto.getGroupId()) || Objects.isNull(batchDto.getDay())) { log.error("消费批量生成用户日统计消息失败，消息格式不正确，消息：{}", message); return; } log.info("开始处理考勤统计批量消息: tenantId={}, groupId={}, 用户数={}, 日期={}", batchDto.getTenantId(), batchDto.getGroupId(), batchDto.getUserIdList().size(), batchDto.getDay()); // 4. 异步延迟执行 CompletableFuture.delayedExecutor(2, TimeUnit.SECONDS, threadPoolExecutor).execute(() -> { try { // 5. 调用业务服务执行统计计算 attendanceDayStatisticsService.batchStatisticChange(batchDto); log.info("考勤统计批量处理完成: tenantId={}, 用户数={}", batchDto.getTenantId(), batchDto.getUserIdList().size()); } catch (Exception ex) { log.error("批量生成用户日统计消息消费失败，参数:{} 错误:{}", message, ex.getMessage(), ex); throw new RuntimeException("批量生成用户日统计消息消费失败，触发重试", ex); } }); } @Override public void prepareStart(DefaultMQPushConsumer consumer) { // 优化消费参数 consumer.setPullBatchSize(16); // 每次拉取的消息数量 consumer.setConsumeMessageBatchMaxSize(1); // 每次批量消费的消息数 consumer.setPullInterval(3000); // 拉取间隔3秒 consumer.setConsumeTimeout(10 * 60 * 1000); // 消费超时10分钟 // 设置消费起始位置（从上次消费的位置继续） consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET); // 调整消费线程池 consumer.setConsumeThreadMin(2); consumer.setConsumeThreadMax(4); log.info("RocketMQ消费者初始化完成: topic={}, group={}", MessageTopicConstants.ATTENDANCE_STATISTICS_BATCH_TOPIC, MessageTopicConstants.ATTENDANCE_STATISTICS_BATCH_CONSUMER_GROUP); } }

四、线程池配置

package jnpf.attendance.bean; import org.jetbrains.annotations.NotNull; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; /** * 线程池配置类 * 根据任务类型（CPU密集、IO密集、混合）创建合适的线程池。 */ @Configuration public class FtbThreadPoolExecutor { // 队列边界 private static final int QUEUE_CAPACITY = 200; // 线程存活时间 s(秒) private static final long KEEP_ALIVE_TIME = 15L; private static final String THREAD_NAME_PREFIX = "FTB服务-考勤模块"; @Bean(name = "attendanceRuleThreadPool", destroyMethod = "shutdown") public ThreadPoolExecutor attendanceRuleThreadPool() { return createNamedThreadPool("attendance-rule", TaskType.IO_INTENSIVE); } @Bean(name = "cpuIntensiveThreadPool", destroyMethod = "shutdown") public ThreadPoolExecutor cpuIntensiveThreadPool() { return createThreadPool(TaskType.CPU_INTENSIVE); } @Bean(name = "ioIntensiveThreadPool", destroyMethod = "shutdown") public ThreadPoolExecutor ioIntensiveThreadPool() { return createThreadPool(TaskType.IO_INTENSIVE); } @Bean(name = "mixedThreadPool", destroyMethod = "shutdown") public ThreadPoolExecutor mixedThreadPool() { return createThreadPool(TaskType.MIXED); } private ThreadPoolExecutor createNamedThreadPool(String name, TaskType type) { int actualProcessors = Math.min(Runtime.getRuntime().availableProcessors(), 42); int runTime, cpuTime; switch (type) { case CPU_INTENSIVE: runTime = 10; cpuTime = 1; break; case IO_INTENSIVE: runTime = 5; cpuTime = 1; break; case MIXED: default: runTime = 50; cpuTime = 1; break; } int maxPoolSize = calculateMaxPoolSize(actualProcessors, runTime, cpuTime, type); BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(QUEUE_CAPACITY); return new ThreadPoolExecutor( actualProcessors, maxPoolSize, KEEP_ALIVE_TIME, TimeUnit.SECONDS, queue, new ThreadFactory() { private final ThreadFactory defaultFactory = Executors.defaultThreadFactory(); private final AtomicInteger count = new AtomicInteger(1); @Override public Thread newThread(@NotNull Runnable r) { Thread thread = defaultFactory.newThread(r); thread.setName("FTB服务-考勤模块-" + name + "-thread-" + count.getAndIncrement()); return thread; } }, new ThreadPoolExecutor.CallerRunsPolicy() ); } private ThreadPoolExecutor createThreadPool(TaskType type) { // 限制为实际核心数 int actualProcessors = Math.min(Runtime.getRuntime().availableProcessors(), 42); int runTime, cpuTime; switch (type) { case CPU_INTENSIVE: runTime = 10; cpuTime = 1; break; case IO_INTENSIVE: runTime = 5; cpuTime = 1; break; case MIXED: default: runTime = 50; cpuTime = 1; break; } int maxPoolSize = calculateMaxPoolSize(actualProcessors, runTime, cpuTime, type); BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(QUEUE_CAPACITY); return new ThreadPoolExecutor( actualProcessors, maxPoolSize, KEEP_ALIVE_TIME, TimeUnit.SECONDS, queue, new ThreadFactory() { private final ThreadFactory defaultFactory = Executors.defaultThreadFactory(); private final AtomicInteger count = new AtomicInteger(1); @Override public Thread newThread(@NotNull Runnable r) { Thread thread = defaultFactory.newThread(r); thread.setName(THREAD_NAME_PREFIX + "-" + type.name().toLowerCase() + "-thread-" + count.getAndIncrement()); return thread; } }, new ThreadPoolExecutor.CallerRunsPolicy() ); } /** * 根据公式：最佳线程数 = N * (1 + (WT / ST)) * 其中 WT = runTime - cpuTime，即线程等待时间 * * @param availableProcessors CPU核心数 * @param runTime 线程总运行时间（单位可自定，如毫秒） * @param cpuTime 线程CPU计算时间 * @return 最佳线程池大小 */ private int calculateMaxPoolSize(int availableProcessors, int runTime, int cpuTime, TaskType type) { if (cpuTime <= 0) { throw new IllegalArgumentException("参数不合法：cpuTime 应大于 0"); } if (type == TaskType.IO_INTENSIVE) { // I/O 密集型：4 倍核心数，上限 200 return Math.min(availableProcessors * 4, 200); } else if (type == TaskType.CPU_INTENSIVE) { // CPU 密集型：等于核心数 return availableProcessors; } else { // 混合型 long ratio = 1L + (long) runTime / cpuTime; long calculated = (long) availableProcessors * ratio; return (int) Math.min(calculated, 200); } } public enum TaskType { CPU_INTENSIVE, IO_INTENSIVE, MIXED } }

五、业务服务实现

java package jnpf.attendance.service.impl; import jnpf.model.attendance.event.AttendanceStatisticsBatchDto; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import javax.annotation.Resource; import java.util.Date; import java.util.List; /** * 考勤日统计服务实现 */ @Slf4j @Service public class AttendanceDayStatisticsServiceImpl { /** * 处理批量统计消息 * * @param batchDto 统计参数 */ @Transactional(rollbackFor = Exception.class) public void batchStatisticChange(AttendanceStatisticsBatchDto batchDto) { log.info("开始执行考勤统计批量处理: tenantId={}, 用户数={}", batchDto.getTenantId(), batchDto.getUserIdList().size()); // 这里实现实际的考勤统计逻辑 for (String userId : batchDto.getUserIdList()) { // 执行单个用户的考勤统计 calculateUserAttendance(batchDto.getTenantId(), userId, batchDto.getDay(), batchDto.getGroupId()); } log.info("考勤统计批量处理完成: tenantId={}, 用户数={}", batchDto.getTenantId(), batchDto.getUserIdList().size()); } /** * 计算单个用户考勤统计 */ private void calculateUserAttendance(String tenantId, String userId, Date day, String groupId) { // 实现具体的考勤统计计算逻辑 // 1. 查询用户当天的考勤记录 // 2. 根据考勤规则计算工时、迟到、早退等 // 3. 保存统计结果到数据库 } }

六、配置文件

application.yml:

rocketmq: name-server: ${ROCKETMQ_NAME_SERVER:192.168.3.24:30094} producer: group: ${ROCKETMQ_PRODUCER_GROUP:fantaibao-producer-group} send-message-timeout: ${ROCKETMQ_SEND_MESSAGE_TIMEOUT:30000} max-message-size: ${ROCKETMQ_MAX_MESSAGE_SIZE:8388608}

消息主题常量:

package jnpf.constants; public interface MessageTopicConstants { // 考勤统计批量生成 String ATTENDANCE_STATISTICS_BATCH_TOPIC = "attendance-statistics-batch-topic"; String ATTENDANCE_STATISTICS_BATCH_CONSUMER_GROUP = "attendance-statistics-batch-consumer-group"; }

七、Maven依赖

<dependency> <groupId>com.alibaba.cloud</groupId> <artifactId>spring-cloud-starter-stream-rocketmq</artifactId> <version>2022.0.0.0</version> </dependency> <dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>2.0.32</version> </dependency>