springboot 消息队列异步导出

### Spring Boot 实现消息队列驱动的异步导出 Excel/CSV 数据 #### 使用场景分析 当面对大量数据导出的需求时，同步处理方式可能导致长时间等待甚至系统崩溃。为了提高用户体验和系统的稳定性，采用基于消息队列的消息传递机制可以有效缓解这些问题。通过引入消息中间件（如 RocketMQ），能够实现生产者消费者模式下的解耦合架构设计。 #### 技术选型说明 对于此类应用场景而言，选择合适的工具至关重要。RocketMQ 是一款分布式消息服务产品，具备高吞吐量、低延迟以及强大的集群管理能力等特点[^4]。它非常适合用来构建大规模的数据传输管道，并且易于集成至现有的微服务体系之中。 #### 方案概述 整个流程大致分为以下几个部分： - **前端请求接收**：用户发起导出请求后，Web 层接收到该指令并将必要的参数封装成 JSON 对象推送到 MQ 中； - **后台任务执行**：监听特定主题(topic) 的 Consumer 组件负责从队列里拉取消息并启动实际的数据读取与写入过程； - **文件存储上传**：完成后的文档会被保存到临时目录下并通过 FTP 或云盘等方式对外提供访问链接； - **通知反馈机制**：最后再向客户端返回一个包含下载路径的通知邮件或站内信提示用户查看结果。 #### 关键代码片段展示 ##### 生产端配置 (`application.yml`) ```yaml rocketmq: name-server: localhost:9876 producer: group: export-producer-group ``` ##### 发送消息实例 (@RestController) ```java @RestController @RequestMapping("/api/export") public class ExportController { @Autowired private DefaultMQProducer mqProducer; @PostMapping("/startExport") public ResponseEntity<String> startExport(@RequestBody Map<String, Object> params){ String topic = "export_topic"; try { Message message = new Message( topic, JSONObject.toJSONString(params).getBytes(StandardCharsets.UTF_8)); SendResult result = this.mqProducer.send(message); return ResponseEntity.ok("Message sent successfully"); } catch (Exception e) { log.error(e.getMessage(),e); throw new RuntimeException("Failed to send message."); } } } ``` ##### 消费端逻辑 (@Component) ```java @Component @Slf4j @RocketMQTransactionListener public class DataExporterConsumer implements RocketMQPushConsumerSelector { @Override public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) { for(MessageExt msg :msgs){ byte[] body=msg.getBody(); String content=new String(body); // 解析传过来的任务信息... Map<String,Object> taskInfo=JSON.parseObject(content,Map.class); // 执行具体的Excel/CVS生成工作... performDataExport(taskInfo); // 记录日志或其他后续动作... log.info("Finished processing message {}",msg.getMsgId()); } return ConsumeOrderlyStatus.SUCCESS; } private void performDataExport(Map<String,Object> param){ // 这里放置真实的业务逻辑， // 可能涉及到查询数据库、组装数据集、调用第三方API等操作 // 假设已经得到了要写出的数据集合 list List<Map<String,String>> dataList=(List<Map<String,String>>)param.get("data"); Workbook workbook=createWorkbook(dataList); File tempFile=writeToFile(workbook,param.get("fileName").toString()); uploadToCloudStorage(tempFile,param.get("bucketName").toString()); } private Workbook createWorkbook(List<Map<String,String>> data){...} private File writeToFile(Workbook wb ,String fileName){...} private void uploadToCloudStorage(File file ,String bucketName){...} } ```