三、架构设计
3.1 整体架构分层
展示层
数据大屏
报表页面
Excel 导出
PDF 报告
↓
API 层
ReportController
ExportController
DashboardController
↓
服务层
ReportQueryService
RealtimeStatService
ExportService
AggregateService
↓
调度层
XXL-Job Admin
XXL-Job Executor
幂等控制
失败重试
↓
同步层
Canal Server
Binlog 解析
MQ 投递
增量同步
↓
存储层
业务 MySQL
报表 MySQL (分区表)
预聚合表
Redis 计数器
OSS 文件存储
3.2 数据流转全景
业务 MySQL
→ Binlog →
Canal
→ MQ →
同步消费者
↓
报表 MySQL
→ 定时聚合 →
预聚合表
→ 查询 →
报表 API
业务 MySQL
→ 业务事件 →
Redis INCR
→ 轮询 →
实时大屏
预聚合表
→ 流式查询 →
EasyExcel
→ 写入 →
OSS
3.3 预聚合表设计
预聚合表就是"提前算好的答案本"。每天凌晨把前一天的统计数据算好存起来，查询时直接查预聚合表而不是原始表。
-- 订单日聚合表：每天一行记录
CREATE TABLE order_daily_stats (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
stat_date DATE NOT NULL COMMENT '统计日期',
merchant_id BIGINT NOT NULL COMMENT '商户ID',
region_code VARCHAR(16) COMMENT '区域编码',
order_count INT NOT NULL DEFAULT 0 COMMENT '订单数',
order_amount DECIMAL(16,2) NOT NULL DEFAULT 0.00 COMMENT '订单金额',
refund_count INT NOT NULL DEFAULT 0 COMMENT '退款数',
refund_amount DECIMAL(16,2) NOT NULL DEFAULT 0.00 COMMENT '退款金额',
gmt_create DATETIME DEFAULT CURRENT_TIMESTAMP,
gmt_modified DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
UNIQUE KEY uk_date_merchant (stat_date, merchant_id),
KEY idx_date (stat_date),
KEY idx_merchant (merchant_id)
) ENGINE=InnoDB COMMENT='订单日聚合表';
-- 小时聚合表：每小时一行，粒度更细
CREATE TABLE order_hourly_stats (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
stat_date DATE NOT NULL,
stat_hour TINYINT NOT NULL COMMENT '0-23',
merchant_id BIGINT NOT NULL,
order_count INT NOT NULL DEFAULT 0,
order_amount DECIMAL(16,2) NOT NULL DEFAULT 0.00,
UNIQUE KEY uk_date_hour_merchant (stat_date, stat_hour, merchant_id)
) ENGINE=InnoDB COMMENT='订单小时聚合表';
查询对比：查原始表 SELECT SUM(order_count) FROM order_daily_stats WHERE stat_date = '2026-06-13' 扫描 1 行；查原始 order 表扫描 500 万行。性能差距 500 万倍！
3.4 分区表设计
对于必须保留的原始明细数据（比如需要查单笔订单详情），用分区表按月拆分，查询时只扫描对应分区。
-- 按月分区的订单明细表
CREATE TABLE order_detail (
id BIGINT NOT NULL AUTO_INCREMENT,
order_no VARCHAR(64) NOT NULL,
user_id BIGINT NOT NULL,
merchant_id BIGINT NOT NULL,
amount DECIMAL(16,2) NOT NULL,
status TINYINT NOT NULL,
create_time DATETIME NOT NULL,
PRIMARY KEY (id, create_time),
KEY idx_order_no (order_no),
KEY idx_user_id (user_id, create_time),
KEY idx_merchant_time (merchant_id, create_time)
) ENGINE=InnoDB
PARTITION BY RANGE (TO_DAYS(create_time)) (
PARTITION p202601 VALUES LESS THAN (TO_DAYS('2026-02-01')),
PARTITION p202602 VALUES LESS THAN (TO_DAYS('2026-03-01')),
PARTITION p202603 VALUES LESS THAN (TO_DAYS('2026-04-01')),
PARTITION p202604 VALUES LESS THAN (TO_DAYS('2026-05-01')),
PARTITION p202605 VALUES LESS THAN (TO_DAYS('2026-06-01')),
PARTITION p202606 VALUES LESS THAN (TO_DAYS('2026-07-01')),
PARTITION pmax VALUES LESS THAN MAXVALUE
);
分区裁剪：查询 WHERE create_time >= '2026-06-01' 时，MySQL 只扫描 p202606 分区，而不是全部数据。就像翻日历只翻6月那几页，不用从头翻到尾。
四、关键流程
4.1 实时统计 — Redis 计数器 + 定期落库
实时统计的核心是"双写"：每次业务事件发生时，既写 Redis 计数器（给前端实时查），也记录到数据库（给后续聚合用）。
用户下单
→
Redis INCR
→
前端轮询实时数据
Redis 计数器
→ 定时刷新(每分钟) →
报表 MySQL
→ 聚合 →
预聚合表
package com.example.report.service;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.TimeUnit;
/**
* 实时统计服务 — 用 Redis 做计数器
*
* Redis Key 设计：
* stat:order:count:{date} → 当日订单数
* stat:order:amount:{date} → 当日订单总金额（用 INCRBYFLOAT）
* stat:user:new:{date} → 当日新增用户数
*
* TTL 设为 7 天，过期自动清理，防止 Key 无限增长
*/
@Slf4j
@Service
@RequiredArgsConstructor
public class RealtimeStatService {
private final StringRedisTemplate redisTemplate;
private static final DateTimeFormatter DATE_FMT = DateTimeFormatter.ofPattern("yyyyMMdd");
/**
* 订单创建时调用 — 订单数 +1，金额累加
* 就像加油站的跳字表：每来一单就跳一下
*/
public void onOrderCreated(long merchantId, double amount) {
String dateKey = LocalDate.now().format(DATE_FMT);
// 全局订单数 +1
String countKey = "stat:order:count:" + dateKey;
redisTemplate.opsForValue().increment(countKey);
redisTemplate.expire(countKey, 7, TimeUnit.DAYS);
// 全局金额累加
String amountKey = "stat:order:amount:" + dateKey;
redisTemplate.opsForValue().increment(amountKey, String.valueOf(amount));
redisTemplate.expire(amountKey, 7, TimeUnit.DAYS);
// 商户维度：订单数 +1
String merchantCountKey = "stat:merchant:count:" + merchantId + ":" + dateKey;
redisTemplate.opsForValue().increment(merchantCountKey);
redisTemplate.expire(merchantCountKey, 7, TimeUnit.DAYS);
// 商户维度：金额累加
String merchantAmountKey = "stat:merchant:amount:" + merchantId + ":" + dateKey;
redisTemplate.opsForValue().increment(merchantAmountKey, String.valueOf(amount));
redisTemplate.expire(merchantAmountKey, 7, TimeUnit.DAYS);
log.info("实时统计更新: merchantId={}, amount={}, date={}", merchantId, amount, dateKey);
}
/**
* 查询今日实时统计数据
* 大屏每 5 秒轮询一次此接口
*/
public RealtimeStatVO getTodayStats() {
String dateKey = LocalDate.now().format(DATE_FMT);
String countStr = redisTemplate.opsForValue().get("stat:order:count:" + dateKey);
String amountStr = redisTemplate.opsForValue().get("stat:order:amount:" + dateKey);
String newUserStr = redisTemplate.opsForValue().get("stat:user:new:" + dateKey);
return RealtimeStatVO.builder()
.orderCount(parseLong(countStr))
.orderAmount(parseDouble(amountStr))
.newUserCount(parseLong(newUserStr))
.statDate(LocalDate.now())
.build();
}
/**
* 查询某商户今日实时统计
*/
public RealtimeStatVO getMerchantTodayStats(long merchantId) {
String dateKey = LocalDate.now().format(DATE_FMT);
String countStr = redisTemplate.opsForValue().get(
"stat:merchant:count:" + merchantId + ":" + dateKey);
String amountStr = redisTemplate.opsForValue().get(
"stat:merchant:amount:" + merchantId + ":" + dateKey);
return RealtimeStatVO.builder()
.orderCount(parseLong(countStr))
.orderAmount(parseDouble(amountStr))
.statDate(LocalDate.now())
.build();
}
private long parseLong(String val) {
return val == null ? 0L : Long.parseLong(val);
}
private double parseDouble(String val) {
return val == null ? 0.0 : Double.parseDouble(val);
}
}
/**
* Redis 计数器 → 数据库 定时刷新任务
* 每分钟执行一次，把 Redis 当天累计值写入中间表
* 注意：这里用 UPSERT 语义，重复执行不会出错（幂等）
*/
@Slf4j
@Component
@RequiredArgsConstructor
public class RedisCounterFlushTask {
private final StringRedisTemplate redisTemplate;
private final JdbcTemplate jdbcTemplate;
private static final DateTimeFormatter DATE_FMT = DateTimeFormatter.ofPattern("yyyyMMdd");
/**
* 每分钟刷新一次 Redis 计数器到数据库
* 用 INSERT ... ON DUPLICATE KEY UPDATE 实现幂等
*/
@Scheduled(cron = "0 */1 * * * ?")
public void flushToDatabase() {
String dateKey = LocalDate.now().format(DATE_FMT);
String countStr = redisTemplate.opsForValue().get("stat:order:count:" + dateKey);
String amountStr = redisTemplate.opsForValue().get("stat:order:amount:" + dateKey);
if (countStr == null) {
return; // 今天还没数据，跳过
}
String sql = """
INSERT INTO realtime_daily_stats (stat_date, order_count, order_amount, gmt_modified)
VALUES (?, ?, ?, NOW())
ON DUPLICATE KEY UPDATE
order_count = VALUES(order_count),
order_amount = VALUES(order_amount),
gmt_modified = NOW()
""";
jdbcTemplate.update(sql,
LocalDate.now(),
Long.parseLong(countStr),
Double.parseDouble(amountStr != null ? amountStr : "0"));
log.info("Redis计数器刷新到数据库: date={}, count={}, amount={}",
dateKey, countStr, amountStr);
}
}
为什么不用 Redis INCRBYFLOAT 存金额？因为浮点数精度问题！0.1 + 0.2 = 0.30000000000000004。生产环境中，金额用"分"为单位存整数（INCRBY），或者用 INCRBYFLOAT 但展示时做四舍五入。更严格的方案是用 MQ 异步写库，Redis 只存整数计数。
4.2 定时任务 — XXL-Job 分布式调度 + 幂等
为什么不用 @Scheduled？因为在集群环境（3台机器）下，@Scheduled 每台机器都会执行一次——本该跑1次的任务跑了3次，报表数据变成3倍！XXL-Job 就是来解决这个问题的。
XXL-Job Admin
→ 分发任务 →
Executor-1
XXL-Job Admin
✗ 不分发
Executor-2
✗ 不分发
Executor-3
Admin 保证同一时刻只有一个 Executor 执行任务
package com.example.report.job;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;
import java.time.LocalDate;
/**
* 订单日聚合任务 — 每天凌晨 1:00 执行
* 把昨天的 order 表数据聚合到 order_daily_stats
*
* 幂等设计：INSERT ... ON DUPLICATE KEY UPDATE
* 即使任务重复执行，结果也是对的
*/
@Slf4j
@Component
@RequiredArgsConstructor
public class OrderDailyAggregateJob {
private final JdbcTemplate jdbcTemplate;
/**
* XXL-Job 任务入口
* 在 XXL-Job Admin 配置 Cron: 0 0 1 * * ?
*/
@XxlJob("orderDailyAggregateJob")
public void execute() {
LocalDate yesterday = LocalDate.now().minusDays(1);
log.info("开始执行订单日聚合任务, 统计日期: {}", yesterday);
try {
aggregateByDate(yesterday);
log.info("订单日聚合任务执行成功, 统计日期: {}", yesterday);
} catch (Exception e) {
log.error("订单日聚合任务执行失败, 统计日期: {}", yesterday, e);
throw new RuntimeException(e); // 抛异常让 XXL-Job 记录失败
}
}
/**
* 核心聚合逻辑：
* 从 order 明细表聚合到 order_daily_stats 预聚合表
* 用 INSERT ... ON DUPLICATE KEY UPDATE 保证幂等
*/
private void aggregateByDate(LocalDate date) {
String sql = """
INSERT INTO order_daily_stats (stat_date, merchant_id, order_count, order_amount)
SELECT
DATE(create_time) AS stat_date,
merchant_id AS merchant_id,
COUNT(*) AS order_count,
COALESCE(SUM(amount), 0) AS order_amount
FROM `order`
WHERE DATE(create_time) = ?
AND status IN (1, 2, 3) -- 排除已取消的订单
GROUP BY DATE(create_time), merchant_id
ON DUPLICATE KEY UPDATE
order_count = VALUES(order_count),
order_amount = VALUES(order_amount),
gmt_modified = NOW()
""";
int rows = jdbcTemplate.update(sql, date);
log.info("聚合完成: date={}, 影响行数={}", date, rows);
}
}
/**
* 更复杂的幂等方案：分布式锁 + 状态标记
* 适用于不能简单 UPSERT 的场景
*/
@Slf4j
@Component
@RequiredArgsConstructor
public class ReportGenerateJob {
private final StringRedisTemplate redisTemplate;
private final ReportService reportService;
@XxlJob("reportGenerateJob")
public void execute() {
LocalDate yesterday = LocalDate.now().minusDays(1);
String lockKey = "job:report:generate:" + yesterday;
// 1. 分布式锁：同一时刻只有一个实例能执行
Boolean locked = redisTemplate.opsForValue()
.setIfAbsent(lockKey, "1", 30, TimeUnit.MINUTES);
if (!Boolean.TRUE.equals(locked)) {
log.warn("任务已在其他节点执行, 跳过: date={}", yesterday);
return;
}
try {
// 2. 状态标记：检查任务是否已完成
String statusKey = "job:report:status:" + yesterday;
String status = redisTemplate.opsForValue().get(statusKey);
if ("SUCCESS".equals(status)) {
log.info("任务已成功执行过, 跳过: date={}", yesterday);
return;
}
// 3. 执行业务逻辑
reportService.generateDailyReport(yesterday);
// 4. 标记成功（保留 7 天）
redisTemplate.opsForValue().set(statusKey, "SUCCESS", 7, TimeUnit.DAYS);
} finally {
// 5. 释放锁
redisTemplate.delete(lockKey);
}
}
}
XXL-Job vs @Scheduled 对比：
@Scheduled：单机调度，集群每台都跑，无失败重试，无可视化管理
XXL-Job：分布式调度只跑一次，自动失败重试，有 Web 管理界面，支持动态调整 Cron、手动触发、查看执行日志
4.3 报表导出 — EasyExcel 流式写入 + OSS
导出50万行数据到 Excel，如果用 Apache POI 全部加载到内存再写入，大约需要 2-3GB 内存——直接 OOM。EasyExcel 的流式写入只在内存中保持一行数据，无论导出多少行，内存都稳定在 20MB 左右。
用户点击导出
→
创建异步导出任务
→
流式查询 DB
逐行查询
→
EasyExcel 逐行写入
→
上传 OSS
OSS 链接
→
通知用户下载
package com.example.report.service;
import com.alibaba.excel.EasyExcel;
import com.alibaba.excel.write.style.column.LongestMatchColumnWidthStyleStrategy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.time.LocalDate;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* 报表导出服务
*
* 核心思路：
* 1. 不一次性查出所有数据 → 用分页流式查询
* 2. 不一次性写完 Excel → EasyExcel 逐行写入
* 3. 不阻塞用户等待 → 异步导出 + OSS + 通知
*
* 核心原则：流式处理，分批写入，避免一次性加载全部数据到内存
*/
@Slf4j
@Service
@RequiredArgsConstructor
public class ReportExportService {
private final JdbcTemplate jdbcTemplate;
private final OssService ossService;
private final ExportTaskService exportTaskService;
private static final int PAGE_SIZE = 5000; // 每次查 5000 行
private final ExecutorService asyncExecutor = Executors.newFixedThreadPool(4);
/**
* 异步导出入口
*/
public String asyncExport(ExportRequest request) {
// 1. 创建导出任务记录
String taskId = exportTaskService.createTask(request);
log.info("创建导出任务: taskId={}, request={}", taskId, request);
// 2. 异步执行导出
asyncExecutor.submit(() -> {
try {
exportTaskService.updateStatus(taskId, "RUNNING");
String ossUrl = doExport(taskId, request);
exportTaskService.updateSuccess(taskId, ossUrl);
} catch (Exception e) {
log.error("导出失败: taskId={}", taskId, e);
exportTaskService.updateFailed(taskId, e.getMessage());
}
});
return taskId;
}
/**
* 核心导出逻辑 — 流式查询 + 流式写入
*/
private String doExport(String taskId, ExportRequest request) {
ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
// EasyExcel 写入器 — 自动适配列宽
var writer = EasyExcel.write(outputStream, OrderExportDTO.class)
.registerWriteHandler(new LongestMatchColumnWidthStyleStrategy())
.build();
var sheet = EasyExcel.writerSheet("订单数据").build();
try {
int page = 0;
int totalRows = 0;
while (true) {
// 分页查询 — 每次只查 PAGE_SIZE 行
String sql = """
SELECT order_no, user_id, merchant_id, amount, status, create_time
FROM `order`
WHERE create_time >= ? AND create_time < ?
ORDER BY id
LIMIT ? OFFSET ?
""";
int offset = page * PAGE_SIZE;
List<OrderExportDTO> rows = jdbcTemplate.query(
sql,
(rs, rowNum) -> OrderExportDTO.builder()
.orderNo(rs.getString("order_no"))
.userId(rs.getLong("user_id"))
.merchantId(rs.getLong("merchant_id"))
.amount(rs.getBigDecimal("amount"))
.status(rs.getInt("status"))
.createTime(rs.getString("create_time"))
.build(),
request.getStartDate(),
request.getEndDate(),
PAGE_SIZE,
offset
);
if (rows.isEmpty()) {
break; // 没有更多数据了
}
// 逐批写入 Excel
writer.write(rows, sheet);
totalRows += rows.size();
page++;
log.info("导出进度: taskId={}, 已写入{}行", taskId, totalRows);
}
} finally {
writer.finish();
}
// 上传到 OSS
String fileName = "order_export_" + taskId + ".xlsx";
String ossUrl = ossService.upload(
fileName,
new ByteArrayInputStream(outputStream.toByteArray()),
outputStream.size()
);
log.info("导出完成: taskId={}, 总行数={}, ossUrl={}", taskId, totalRows, ossUrl);
return ossUrl;
}
}
/**
* 导出 DTO — EasyExcel 注解定义列名和格式
*/
import com.alibaba.excel.annotation.ExcelProperty;
import com.alibaba.excel.annotation.format.NumberFormat;
import lombok.Builder;
import lombok.Data;
@Data
@Builder
public class OrderExportDTO {
@ExcelProperty("订单号")
private String orderNo;
@ExcelProperty("用户ID")
private Long userId;
@ExcelProperty("商户ID")
private Long merchantId;
@ExcelProperty("订单金额")
@NumberFormat("#,##0.00")
private BigDecimal amount;
@ExcelProperty("状态")
private Integer status;
@ExcelProperty("创建时间")
private String createTime;
}
/**
* 导出任务服务 — 记录导出状态，供前端轮询
*/
@Service
@RequiredArgsConstructor
public class ExportTaskService {
private final ExportTaskMapper exportTaskMapper;
public String createTask(ExportRequest request) {
ExportTask task = new ExportTask();
task.setTaskId(UUID.randomUUID().toString());
task.setStatus("CREATED");
task.setRequest(JSON.toJSONString(request));
task.setCreateTime(new Date());
exportTaskMapper.insert(task);
return task.getTaskId();
}
public void updateStatus(String taskId, String status) {
exportTaskMapper.updateStatus(taskId, status);
}
public void updateSuccess(String taskId, String ossUrl) {
exportTaskMapper.updateSuccess(taskId, "SUCCESS", ossUrl, new Date());
}
public void updateFailed(String taskId, String errorMsg) {
exportTaskMapper.updateFailed(taskId, "FAILED", errorMsg, new Date());
}
/**
* 前端轮询接口：查询导出任务状态
* 返回 CREATED / RUNNING / SUCCESS / FAILED
*/
public ExportTaskVO getTaskStatus(String taskId) {
ExportTask task = exportTaskMapper.selectByTaskId(taskId);
if (task == null) {
throw new BizException("任务不存在");
}
return ExportTaskVO.builder()
.taskId(task.getTaskId())
.status(task.getStatus())
.downloadUrl(task.getOssUrl())
.errorMsg(task.getErrorMsg())
.build();
}
}
内存对比：POI 全量加载 50 万行 ≈ 2GB 内存；EasyExcel 流式写入 50 万行 ≈ 20MB 内存。差了 100 倍！就像一个是把整个游泳池的水灌进杯子，一个是用管子慢慢流。
4.4 数据同步 — Canal 监听 Binlog
Canal 是阿里开源的 MySQL Binlog 增量订阅组件。它伪装成 MySQL 的从库，接收 Binlog 事件，然后投递到 MQ（RocketMQ/Kafka），消费者拿到变更数据写入报表库。
业务 MySQL (Master)
→ Binlog →
Canal (伪装为 Slave)
Canal
→ 投递 →
RocketMQ
→ 消费 →
同步消费者
同步消费者
→ 写入 →
报表 MySQL
/**
* Canal 数据同步消费者
* 监听 MQ 中的 Binlog 变更事件，同步到报表库
*
* Canal 投递的消息格式（简化）：
* {
* "type": "INSERT", // INSERT / UPDATE / DELETE
* "table": "order",
* "data": [{ // 变更后的数据
* "id": "123456",
* "order_no": "ORD20260613001",
* "amount": "99.90",
* ...
* }],
* "old": [{ // 变更前的数据 (UPDATE 时有)
* "status": "1"
* }]
* }
*/
@Slf4j
@Component
@RequiredArgsConstructor
public class CanalSyncConsumer {
private final JdbcTemplate reportJdbcTemplate;
private final RedisTemplate<String, String> redisTemplate;
/**
* 消费 Canal 同步消息
*/
@RocketMQMessageListener(
topic = "canal-topic",
consumerGroup = "canal-sync-group"
)
public void onMessage(CanalMessage message) {
log.info("收到Canal同步消息: type={}, table={}",
message.getType(), message.getTable());
switch (message.getTable()) {
case "order" -> syncOrder(message);
case "user" -> syncUser(message);
case "merchant" -> syncMerchant(message);
default -> log.warn("未处理的表: {}", message.getTable());
}
}
/**
* 同步订单数据到报表库
*/
private void syncOrder(CanalMessage message) {
switch (message.getType()) {
case "INSERT" -> {
for (Map<String, String> data : message.getData()) {
String sql = """
INSERT INTO report_order (id, order_no, user_id, merchant_id,
amount, status, create_time)
VALUES (?, ?, ?, ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE
status = VALUES(status),
amount = VALUES(amount)
""";
reportJdbcTemplate.update(sql,
toLong(data.get("id")),
data.get("order_no"),
toLong(data.get("user_id")),
toLong(data.get("merchant_id")),
toDecimal(data.get("amount")),
toInt(data.get("status")),
data.get("create_time")
);
}
}
case "UPDATE" -> {
for (int i = 0; i < message.getData().size(); i++) {
Map<String, String> newData = message.getData().get(i);
Map<String, String> oldData = message.getOld().get(i);
// 只更新变化的字段
StringBuilder sql = new StringBuilder("UPDATE report_order SET ");
List<Object> params = new ArrayList<>();
if (oldData.containsKey("status")) {
sql.append("status = ?, ");
params.add(toInt(newData.get("status")));
}
if (oldData.containsKey("amount")) {
sql.append("amount = ?, ");
params.add(toDecimal(newData.get("amount")));
}
if (!params.isEmpty()) {
sql.append("gmt_modified = NOW() WHERE id = ?");
params.add(toLong(newData.get("id")));
reportJdbcTemplate.update(sql.toString(), params.toArray());
}
}
}
case "DELETE" -> {
for (Map<String, String> data : message.getData()) {
reportJdbcTemplate.update(
"DELETE FROM report_order WHERE id = ?",
toLong(data.get("id")));
}
}
}
}
private long toLong(String val) { return val == null ? 0L : Long.parseLong(val); }
private int toInt(String val) { return val == null ? 0 : Integer.parseInt(val); }
private BigDecimal toDecimal(String val) { return val == null ? BigDecimal.ZERO : new BigDecimal(val); }
}
Canal 配置要点：
1. MySQL 开启 Binlog：log-bin=mysql-bin，binlog-format=ROW
2. 创建 Canal 账号：GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%'
3. Canal 的 canal.instance.filter.regex 配置只监听需要的表，避免全量同步
4. 消费端做好幂等（用 ON DUPLICATE KEY UPDATE），防止重复消费
4.5 聚合引擎 — 完整定时聚合服务
/**
* 聚合引擎 — 将明细数据定期聚合到预聚合表
*
* 聚合层级：
* 小时聚合 (每小时) → order_hourly_stats
* 日聚合 (每天) → order_daily_stats
* 月聚合 (每月) → order_monthly_stats
*
* 设计原则：
* 1. 所有聚合操作都用 INSERT ... ON DUPLICATE KEY UPDATE 保证幂等
* 2. 支持手动重跑某个日期的聚合（补偿机制）
* 3. 聚合前先标记，聚合后校验数据量
*/
@Slf4j
@Component
@RequiredArgsConstructor
public class AggregateEngine {
private final JdbcTemplate reportJdbcTemplate;
/**
* 小时聚合任务 — 每小时执行
* Cron: 0 5 * * * ? (每小时的第5分钟执行，留出5分钟延迟窗口)
*/
@XxlJob("orderHourlyAggregateJob")
public void hourlyAggregate() {
// 聚合上一小时的数据
int hour = LocalDateTime.now().minusHours(1).getHour();
LocalDate date = LocalDate.now().minusHours(1);
String sql = """
INSERT INTO order_hourly_stats (stat_date, stat_hour, merchant_id, order_count, order_amount)
SELECT
DATE(create_time),
HOUR(create_time),
merchant_id,
COUNT(*),
COALESCE(SUM(amount), 0)
FROM report_order
WHERE DATE(create_time) = ? AND HOUR(create_time) = ?
GROUP BY DATE(create_time), HOUR(create_time), merchant_id
ON DUPLICATE KEY UPDATE
order_count = VALUES(order_count),
order_amount = VALUES(order_amount)
""";
int rows = reportJdbcTemplate.update(sql, date, hour);
log.info("小时聚合完成: date={}, hour={}, 影响行数={}", date, hour, rows);
}
/**
* 日聚合任务 — 每天凌晨 2:00 执行
* 可以从小时聚合表汇总，也可以直接从明细表聚合
* 这里选择直接从明细表聚合，更准确
*/
@XxlJob("orderDailyAggregateJob")
public void dailyAggregate() {
LocalDate yesterday = LocalDate.now().minusDays(1);
aggregateByDate(yesterday);
}
/**
* 手动补偿 — 重跑某个日期的聚合
* 在 XXL-Job Admin 手动触发时传入日期参数
*/
@XxlJob("orderDailyAggregateCompensateJob")
public void dailyAggregateCompensate() {
String param = XxlJobHelper.getJobParam(); // 从 Admin 传来的参数
if (StringUtils.isBlank(param)) {
log.error("补偿任务缺少日期参数, 格式: 2026-06-12");
return;
}
LocalDate date = LocalDate.parse(param);
aggregateByDate(date);
log.info("补偿聚合完成: date={}", date);
}
private void aggregateByDate(LocalDate date) {
String sql = """
INSERT INTO order_daily_stats (stat_date, merchant_id, region_code, order_count, order_amount)
SELECT
DATE(create_time),
merchant_id,
region_code,
COUNT(*),
COALESCE(SUM(amount), 0)
FROM report_order
WHERE DATE(create_time) = ?
GROUP BY DATE(create_time), merchant_id, region_code
ON DUPLICATE KEY UPDATE
order_count = VALUES(order_count),
order_amount = VALUES(order_amount),
gmt_modified = NOW()
""";
int rows = reportJdbcTemplate.update(sql, date);
log.info("日聚合完成: date={}, 影响行数={}", date, rows);
}
/**
* 月聚合任务 — 每月1号凌晨 3:00 执行
* 从日聚合表汇总，速度更快
*/
@XxlJob("orderMonthlyAggregateJob")
public void monthlyAggregate() {
YearMonth lastMonth = YearMonth.now().minusMonths(1);
String sql = """
INSERT INTO order_monthly_stats (stat_month, merchant_id, order_count, order_amount)
SELECT
DATE_FORMAT(stat_date, '%%Y-%%m'),
merchant_id,
SUM(order_count),
SUM(order_amount)
FROM order_daily_stats
WHERE DATE_FORMAT(stat_date, '%%Y-%%m') = ?
GROUP BY DATE_FORMAT(stat_date, '%%Y-%%m'), merchant_id
ON DUPLICATE KEY UPDATE
order_count = VALUES(order_count),
order_amount = VALUES(order_amount)
""";
int rows = reportJdbcTemplate.update(sql, lastMonth.toString());
log.info("月聚合完成: month={}, 影响行数={}", lastMonth, rows);
}
}
4.6 报表查询 API — 查预聚合表而非明细表
@RestController
@RequestMapping("/api/report")
@RequiredArgsConstructor
public class ReportController {
private final RealtimeStatService realtimeStatService;
private final ReportQueryService reportQueryService;
private final ReportExportService reportExportService;
/**
* 实时统计 — 大屏轮询接口
* 直接查 Redis 计数器，不走数据库
*/
@GetMapping("/realtime")
public Result<RealtimeStatVO> realtime() {
return Result.ok(realtimeStatService.getTodayStats());
}
/**
* 日报表查询 — 查预聚合表
* 支持：按日期范围、按商户、按区域查询
*/
@GetMapping("/daily")
public Result<PageResult<DailyReportVO>> dailyReport(
@RequestParam LocalDate startDate,
@RequestParam LocalDate endDate,
@RequestParam(required = false) Long merchantId,
@RequestParam(required = false) String regionCode,
@RequestParam(defaultValue = "1") int page,
@RequestParam(defaultValue = "20") int size) {
return Result.ok(reportQueryService.queryDailyReport(
startDate, endDate, merchantId, regionCode, page, size));
}
/**
* 趋势图数据 — 按天聚合
* 返回日期 → 订单数/金额 的时序数据
*/
@GetMapping("/trend")
public Result<List<TrendVO>> trend(
@RequestParam LocalDate startDate,
@RequestParam LocalDate endDate,
@RequestParam(required = false) Long merchantId) {
return Result.ok(reportQueryService.queryTrend(startDate, endDate, merchantId));
}
/**
* 异步导出 — 返回任务ID
*/
@PostMapping("/export")
public Result<String> export(@RequestBody ExportRequest request) {
String taskId = reportExportService.asyncExport(request);
return Result.ok(taskId);
}
/**
* 查询导出任务状态
*/
@GetMapping("/export/{taskId}")
public Result<ExportTaskVO> exportStatus(@PathVariable String taskId) {
return Result.ok(exportTaskService.getTaskStatus(taskId));
}
}
/**
* 报表查询服务 — 查预聚合表，永远不直接查明细表
*/
@Service
@RequiredArgsConstructor
public class ReportQueryService {
private final JdbcTemplate reportJdbcTemplate;
public PageResult<DailyReportVO> queryDailyReport(
LocalDate startDate, LocalDate endDate,
Long merchantId, String regionCode,
int page, int size) {
StringBuilder sql = new StringBuilder("""
SELECT stat_date, merchant_id, region_code,
order_count, order_amount
FROM order_daily_stats
WHERE stat_date BETWEEN ? AND ?
""");
List<Object> params = new ArrayList<>();
params.add(startDate);
params.add(endDate);
if (merchantId != null) {
sql.append(" AND merchant_id = ?");
params.add(merchantId);
}
if (regionCode != null) {
sql.append(" AND region_code = ?");
params.add(regionCode);
}
sql.append(" ORDER BY stat_date DESC");
sql.append(" LIMIT ? OFFSET ?");
params.add(size);
params.add((page - 1) * size);
List<DailyReportVO> list = reportJdbcTemplate.query(
sql.toString(),
(rs, rowNum) -> DailyReportVO.builder()
.statDate(rs.getDate("stat_date").toLocalDate())
.merchantId(rs.getLong("merchant_id"))
.regionCode(rs.getString("region_code"))
.orderCount(rs.getInt("order_count"))
.orderAmount(rs.getBigDecimal("order_amount"))
.build(),
params.toArray()
);
return new PageResult<>(list, page, size);
}
/**
* 趋势数据 — 按天汇总
*/
public List<TrendVO> queryTrend(LocalDate startDate, LocalDate endDate, Long merchantId) {
StringBuilder sql = new StringBuilder("""
SELECT stat_date,
SUM(order_count) AS total_count,
SUM(order_amount) AS total_amount
FROM order_daily_stats
WHERE stat_date BETWEEN ? AND ?
""");
List<Object> params = new ArrayList<>();
params.add(startDate);
params.add(endDate);
if (merchantId != null) {
sql.append(" AND merchant_id = ?");
params.add(merchantId);
}
sql.append(" GROUP BY stat_date ORDER BY stat_date");
return reportJdbcTemplate.query(sql.toString(), (rs, rowNum) -> TrendVO.builder()
.date(rs.getDate("stat_date").toLocalDate())
.orderCount(rs.getInt("total_count"))
.orderAmount(rs.getBigDecimal("total_amount"))
.build(), params.toArray());
}
}
4.7 Canal 高可用与增量补偿
Canal 宕机或 MQ 消费延迟怎么办？需要一个全量补偿机制来兜底。
/**
* 数据一致性校验与补偿任务
* 每天凌晨 4:00 执行
* 对比业务库和报表库的数据量，发现差异则重跑聚合
*/
@Slf4j
@Component
@RequiredArgsConstructor
public class DataConsistencyCheckJob {
private final JdbcTemplate bizJdbcTemplate; // 业务库
private final JdbcTemplate reportJdbcTemplate; // 报表库
private final AggregateEngine aggregateEngine;
@XxlJob("dataConsistencyCheckJob")
public void check() {
LocalDate yesterday = LocalDate.now().minusDays(1);
// 1. 查业务库的原始订单数
Long bizCount = bizJdbcTemplate.queryForObject(
"SELECT COUNT(*) FROM `order` WHERE DATE(create_time) = ?",
Long.class, yesterday);
// 2. 查报表库聚合表的订单数
Long reportCount = reportJdbcTemplate.queryForObject(
"SELECT COALESCE(SUM(order_count), 0) FROM order_daily_stats WHERE stat_date = ?",
Long.class, yesterday);
log.info("数据一致性校验: date={}, 业务库={}, 报表库={}", yesterday, bizCount, reportCount);
// 3. 如果不一致，触发补偿聚合
if (bizCount != null && reportCount != null
&& !bizCount.equals(reportCount)) {
log.warn("数据不一致! 触发补偿聚合: date={}, 差异={}",
yesterday, bizCount - reportCount);
aggregateEngine.aggregateByDate(yesterday);
// 4. 补偿后再校验一次
Long newReportCount = reportJdbcTemplate.queryForObject(
"SELECT COALESCE(SUM(order_count), 0) FROM order_daily_stats WHERE stat_date = ?",
Long.class, yesterday);
if (bizCount.equals(newReportCount)) {
log.info("补偿成功! 数据已一致: date={}, count={}", yesterday, bizCount);
} else {
log.error("补偿后仍不一致! date={}, 业务库={}, 补偿后报表库={}",
yesterday, bizCount, newReportCount);
// 发送告警通知
}
}
}
}
Canal 高可用方案：
1. Canal Server 集群部署（多个实例，Zookeeper 选主）
2. MQ 削峰填谷（Canal → MQ → 消费者），避免数据库被打爆
3. 消费端幂等（ON DUPLICATE KEY UPDATE），MQ 重复投递不会出错
4. 定时全量校验兜底，确保最终一致性