第149集批量写MySQL减少Maxwell事件数Java实战

/**
 * 批量写MySQL配置类
 * @author 运维实战
 */
@Configuration
@EnableConfigurationProperties(MySQLBatchWriteProperties.class)
public class MySQLBatchWriteConfig {
    
    @Autowired
    private MySQLBatchWriteProperties properties;
    
    /**
     * 批量写MySQL服务
     * @return 批量写服务
     */
    @Bean
    public MySQLBatchWriteService mysqlBatchWriteService() {
        return new MySQLBatchWriteService();
    }
    
    /**
     * Maxwell事件监控服务
     * @return 事件监控服务
     */
    @Bean
    public MaxwellEventMonitorService maxwellEventMonitorService() {
        return new MaxwellEventMonitorService();
    }
    
    /**
     * 批量写策略服务
     * @return 策略服务
     */
    @Bean
    public BatchWriteStrategyService batchWriteStrategyService() {
        return new BatchWriteStrategyService();
    }
    
    private static final Logger logger = LoggerFactory.getLogger(MySQLBatchWriteConfig.class);
}

/**
 * 批量写MySQL属性配置
 * @author 运维实战
 */
@Data
@ConfigurationProperties(prefix = "mysql.batch.write")
public class MySQLBatchWriteProperties {
    
    /**
     * 最大批量写入数量
     */
    private int maxBatchSize = 1000;
    
    /**
     * 批量写入超时时间（毫秒）
     */
    private long batchTimeoutMs = 5000;
    
    /**
     * 是否启用批量写入
     */
    private boolean enableBatchWrite = true;
    
    /**
     * 是否启用Maxwell事件优化
     */
    private boolean enableMaxwellOptimization = true;
    
    /**
     * Maxwell事件减少阈值
     */
    private double maxwellEventReductionThreshold = 0.5;
    
    /**
     * 是否启用事务批量写入
     */
    private boolean enableTransactionalBatch = true;
    
    /**
     * 事务超时时间（毫秒）
     */
    private long transactionTimeoutMs = 30000;
    
    /**
     * 是否启用批量插入
     */
    private boolean enableBatchInsert = true;
    
    /**
     * 是否启用批量更新
     */
    private boolean enableBatchUpdate = true;
    
    /**
     * 是否启用批量删除
     */
    private boolean enableBatchDelete = true;
    
    /**
     * 是否启用监控
     */
    private boolean enableMonitor = true;
    
    /**
     * 监控间隔（毫秒）
     */
    private long monitorInterval = 30000;
}

/**
 * 基础批量写MySQL服务
 * @author 运维实战
 */
@Service
public class MySQLBatchWriteService {
    
    @Autowired
    private JdbcTemplate jdbcTemplate;
    
    @Autowired
    private MySQLBatchWriteProperties properties;
    
    @Autowired
    private MaxwellEventMonitorService maxwellEventMonitorService;
    
    @Autowired
    private BatchWriteStrategyService batchWriteStrategyService;
    
    private static final Logger logger = LoggerFactory.getLogger(MySQLBatchWriteService.class);
    
    /**
     * 批量插入减少Maxwell事件数
     * @param tableName 表名
     * @param dataList 数据列表
     * @return 批量写入结果
     */
    public MySQLBatchWriteResult batchInsertWithMaxwellOptimization(String tableName, List<Map<String, Object>> dataList) {
        logger.info("开始批量插入操作，表名: {}, 数据量: {}", tableName, dataList.size());
        
        MySQLBatchWriteResult result = new MySQLBatchWriteResult();
        result.setTableName(tableName);
        result.setTotalCount(dataList.size());
        result.setStartTime(System.currentTimeMillis());
        
        try {
            // 获取批量写入策略
            BatchWriteStrategy strategy = batchWriteStrategyService.getStrategy(dataList.size());
            
            // 执行批量插入
            result = executeBatchInsert(tableName, dataList, strategy);
            
            // 记录Maxwell事件优化指标
            maxwellEventMonitorService.recordBatchInsert(tableName, dataList.size(), result.getSuccessCount());
            
            logger.info("批量插入完成，表名: {}, 成功: {}, 失败: {}, 总耗时: {}ms", 
                tableName, result.getSuccessCount(), result.getFailureCount(), result.getDuration());
            
            return result;
            
        } catch (Exception e) {
            logger.error("批量插入异常，表名: {}", tableName, e);
            result.setSuccess(false);
            result.setError("批量插入异常: " + e.getMessage());
            result.setEndTime(System.currentTimeMillis());
            return result;
        }
    }
    
    /**
     * 批量更新减少Maxwell事件数
     * @param tableName 表名
     * @param updateDataList 更新数据列表
     * @return 批量写入结果
     */
    public MySQLBatchWriteResult batchUpdateWithMaxwellOptimization(String tableName, List<Map<String, Object>> updateDataList) {
        logger.info("开始批量更新操作，表名: {}, 数据量: {}", tableName, updateDataList.size());
        
        MySQLBatchWriteResult result = new MySQLBatchWriteResult();
        result.setTableName(tableName);
        result.setTotalCount(updateDataList.size());
        result.setStartTime(System.currentTimeMillis());
        
        try {
            // 获取批量写入策略
            BatchWriteStrategy strategy = batchWriteStrategyService.getStrategy(updateDataList.size());
            
            // 执行批量更新
            result = executeBatchUpdate(tableName, updateDataList, strategy);
            
            // 记录Maxwell事件优化指标
            maxwellEventMonitorService.recordBatchUpdate(tableName, updateDataList.size(), result.getSuccessCount());
            
            logger.info("批量更新完成，表名: {}, 成功: {}, 失败: {}, 总耗时: {}ms", 
                tableName, result.getSuccessCount(), result.getFailureCount(), result.getDuration());
            
            return result;
            
        } catch (Exception e) {
            logger.error("批量更新异常，表名: {}", tableName, e);
            result.setSuccess(false);
            result.setError("批量更新异常: " + e.getMessage());
            result.setEndTime(System.currentTimeMillis());
            return result;
        }
    }
    
    /**
     * 批量删除减少Maxwell事件数
     * @param tableName 表名
     * @param deleteDataList 删除数据列表
     * @return 批量写入结果
     */
    public MySQLBatchWriteResult batchDeleteWithMaxwellOptimization(String tableName, List<Map<String, Object>> deleteDataList) {
        logger.info("开始批量删除操作，表名: {}, 数据量: {}", tableName, deleteDataList.size());
        
        MySQLBatchWriteResult result = new MySQLBatchWriteResult();
        result.setTableName(tableName);
        result.setTotalCount(deleteDataList.size());
        result.setStartTime(System.currentTimeMillis());
        
        try {
            // 获取批量写入策略
            BatchWriteStrategy strategy = batchWriteStrategyService.getStrategy(deleteDataList.size());
            
            // 执行批量删除
            result = executeBatchDelete(tableName, deleteDataList, strategy);
            
            // 记录Maxwell事件优化指标
            maxwellEventMonitorService.recordBatchDelete(tableName, deleteDataList.size(), result.getSuccessCount());
            
            logger.info("批量删除完成，表名: {}, 成功: {}, 失败: {}, 总耗时: {}ms", 
                tableName, result.getSuccessCount(), result.getFailureCount(), result.getDuration());
            
            return result;
            
        } catch (Exception e) {
            logger.error("批量删除异常，表名: {}", tableName, e);
            result.setSuccess(false);
            result.setError("批量删除异常: " + e.getMessage());
            result.setEndTime(System.currentTimeMillis());
            return result;
        }
    }
    
    /**
     * 执行批量插入
     * @param tableName 表名
     * @param dataList 数据列表
     * @param strategy 批量写入策略
     * @return 批量写入结果
     */
    private MySQLBatchWriteResult executeBatchInsert(String tableName, List<Map<String, Object>> dataList, BatchWriteStrategy strategy) {
        MySQLBatchWriteResult result = new MySQLBatchWriteResult();
        result.setTableName(tableName);
        result.setTotalCount(dataList.size());
        result.setStartTime(System.currentTimeMillis());
        
        try {
            if (properties.isEnableTransactionalBatch()) {
                // 使用事务批量插入
                result = executeTransactionalBatchInsert(tableName, dataList, strategy);
            } else {
                // 使用普通批量插入
                result = executeNormalBatchInsert(tableName, dataList, strategy);
            }
            
            result.setEndTime(System.currentTimeMillis());
            return result;
            
        } catch (Exception e) {
            logger.error("执行批量插入异常，表名: {}", tableName, e);
            result.setSuccess(false);
            result.setError("执行批量插入异常: " + e.getMessage());
            result.setEndTime(System.currentTimeMillis());
            return result;
        }
    }
    
    /**
     * 执行事务批量插入
     * @param tableName 表名
     * @param dataList 数据列表
     * @param strategy 批量写入策略
     * @return 批量写入结果
     */
    private MySQLBatchWriteResult executeTransactionalBatchInsert(String tableName, List<Map<String, Object>> dataList, BatchWriteStrategy strategy) {
        MySQLBatchWriteResult result = new MySQLBatchWriteResult();
        result.setTableName(tableName);
        result.setTotalCount(dataList.size());
        result.setStartTime(System.currentTimeMillis());
        
        TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
        transactionTemplate.setTimeout((int) (properties.getTransactionTimeoutMs() / 1000));
        
        try {
            Integer successCount = transactionTemplate.execute(status -> {
                try {
                    // 构建批量插入SQL
                    String sql = buildBatchInsertSQL(tableName, dataList);
                    
                    // 执行批量插入
                    int[] results = jdbcTemplate.batchUpdate(sql, new BatchPreparedStatementSetter() {
                        @Override
                        public void setValues(PreparedStatement ps, int i) throws SQLException {
                            Map<String, Object> data = dataList.get(i);
                            int paramIndex = 1;
                            
                            for (Map.Entry<String, Object> entry : data.entrySet()) {
                                ps.setObject(paramIndex++, entry.getValue());
                            }
                        }
                        
                        @Override
                        public int getBatchSize() {
                            return dataList.size();
                        }
                    });
                    
                    // 计算成功数量
                    int success = 0;
                    for (int res : results) {
                        if (res > 0) {
                            success++;
                        }
                    }
                    
                    return success;
                    
                } catch (Exception e) {
                    status.setRollbackOnly();
                    throw new RuntimeException("事务批量插入失败", e);
                }
            });
            
            result.setSuccessCount(successCount);
            result.setFailureCount(dataList.size() - successCount);
            result.setSuccess(successCount > 0);
            result.setEndTime(System.currentTimeMillis());
            
            return result;
            
        } catch (Exception e) {
            logger.error("事务批量插入异常，表名: {}", tableName, e);
            result.setSuccess(false);
            result.setError("事务批量插入异常: " + e.getMessage());
            result.setEndTime(System.currentTimeMillis());
            return result;
        }
    }
    
    /**
     * 执行普通批量插入
     * @param tableName 表名
     * @param dataList 数据列表
     * @param strategy 批量写入策略
     * @return 批量写入结果
     */
    private MySQLBatchWriteResult executeNormalBatchInsert(String tableName, List<Map<String, Object>> dataList, BatchWriteStrategy strategy) {
        MySQLBatchWriteResult result = new MySQLBatchWriteResult();
        result.setTableName(tableName);
        result.setTotalCount(dataList.size());
        result.setStartTime(System.currentTimeMillis());
        
        try {
            // 构建批量插入SQL
            String sql = buildBatchInsertSQL(tableName, dataList);
            
            // 执行批量插入
            int[] results = jdbcTemplate.batchUpdate(sql, new BatchPreparedStatementSetter() {
                @Override
                public void setValues(PreparedStatement ps, int i) throws SQLException {
                    Map<String, Object> data = dataList.get(i);
                    int paramIndex = 1;
                    
                    for (Map.Entry<String, Object> entry : data.entrySet()) {
                        ps.setObject(paramIndex++, entry.getValue());
                    }
                }
                
                @Override
                public int getBatchSize() {
                    return dataList.size();
                }
            });
            
            // 计算成功数量
            int successCount = 0;
            for (int res : results) {
                if (res > 0) {
                    successCount++;
                }
            }
            
            result.setSuccessCount(successCount);
            result.setFailureCount(dataList.size() - successCount);
            result.setSuccess(successCount > 0);
            result.setEndTime(System.currentTimeMillis());
            
            return result;
            
        } catch (Exception e) {
            logger.error("普通批量插入异常，表名: {}", tableName, e);
            result.setSuccess(false);
            result.setError("普通批量插入异常: " + e.getMessage());
            result.setEndTime(System.currentTimeMillis());
            return result;
        }
    }
    
    /**
     * 构建批量插入SQL
     * @param tableName 表名
     * @param dataList 数据列表
     * @return SQL语句
     */
    private String buildBatchInsertSQL(String tableName, List<Map<String, Object>> dataList) {
        if (dataList.isEmpty()) {
            throw new IllegalArgumentException("数据列表不能为空");
        }
        
        Map<String, Object> firstData = dataList.get(0);
        StringBuilder sql = new StringBuilder();
        sql.append("INSERT INTO ").append(tableName).append(" (");
        
        // 构建列名
        StringJoiner columns = new StringJoiner(", ");
        for (String column : firstData.keySet()) {
            columns.add(column);
        }
        sql.append(columns.toString()).append(") VALUES (");
        
        // 构建占位符
        StringJoiner placeholders = new StringJoiner(", ");
        for (int i = 0; i < firstData.size(); i++) {
            placeholders.add("?");
        }
        sql.append(placeholders.toString()).append(")");
        
        return sql.toString();
    }
    
    /**
     * 执行批量更新
     * @param tableName 表名
     * @param updateDataList 更新数据列表
     * @param strategy 批量写入策略
     * @return 批量写入结果
     */
    private MySQLBatchWriteResult executeBatchUpdate(String tableName, List<Map<String, Object>> updateDataList, BatchWriteStrategy strategy) {
        // 实现批量更新逻辑
        MySQLBatchWriteResult result = new MySQLBatchWriteResult();
        result.setTableName(tableName);
        result.setTotalCount(updateDataList.size());
        result.setStartTime(System.currentTimeMillis());
        
        try {
            // 模拟批量更新操作
            Thread.sleep(updateDataList.size() * 10);
            
            result.setSuccessCount(updateDataList.size());
            result.setFailureCount(0);
            result.setSuccess(true);
            result.setEndTime(System.currentTimeMillis());
            
            return result;
            
        } catch (Exception e) {
            logger.error("执行批量更新异常，表名: {}", tableName, e);
            result.setSuccess(false);
            result.setError("执行批量更新异常: " + e.getMessage());
            result.setEndTime(System.currentTimeMillis());
            return result;
        }
    }
    
    /**
     * 执行批量删除
     * @param tableName 表名
     * @param deleteDataList 删除数据列表
     * @param strategy 批量写入策略
     * @return 批量写入结果
     */
    private MySQLBatchWriteResult executeBatchDelete(String tableName, List<Map<String, Object>> deleteDataList, BatchWriteStrategy strategy) {
        // 实现批量删除逻辑
        MySQLBatchWriteResult result = new MySQLBatchWriteResult();
        result.setTableName(tableName);
        result.setTotalCount(deleteDataList.size());
        result.setStartTime(System.currentTimeMillis());
        
        try {
            // 模拟批量删除操作
            Thread.sleep(deleteDataList.size() * 8);
            
            result.setSuccessCount(deleteDataList.size());
            result.setFailureCount(0);
            result.setSuccess(true);
            result.setEndTime(System.currentTimeMillis());
            
            return result;
            
        } catch (Exception e) {
            logger.error("执行批量删除异常，表名: {}", tableName, e);
            result.setSuccess(false);
            result.setError("执行批量删除异常: " + e.getMessage());
            result.setEndTime(System.currentTimeMillis());
            return result;
        }
    }
    
    @Autowired
    private PlatformTransactionManager transactionManager;
}

/**
 * MySQL批量写入结果类
 * @author 运维实战
 */
@Data
public class MySQLBatchWriteResult {
    
    private boolean success;
    private String tableName;
    private int totalCount;
    private int successCount;
    private int failureCount;
    private String error;
    private long startTime;
    private long endTime;
    
    public MySQLBatchWriteResult() {
        this.success = false;
        this.successCount = 0;
        this.failureCount = 0;
    }
    
    /**
     * 获取处理耗时
     * @return 处理耗时（毫秒）
     */
    public long getDuration() {
        return endTime - startTime;
    }
    
    /**
     * 获取成功率
     * @return 成功率
     */
    public double getSuccessRate() {
        if (totalCount == 0) return 0.0;
        return (double) successCount / totalCount * 100;
    }
    
    /**
     * 获取失败率
     * @return 失败率
     */
    public double getFailureRate() {
        if (totalCount == 0) return 0.0;
        return (double) failureCount / totalCount * 100;
    }
    
    /**
     * 是否全部成功
     * @return 是否全部成功
     */
    public boolean isAllSuccess() {
        return failureCount == 0;
    }
}

/**
 * 批量写入策略类
 * @author 运维实战
 */
@Data
public class BatchWriteStrategy {
    
    private int batchSize;
    private String strategy;
    private String description;
    private long estimatedDuration;
    private long estimatedMaxwellEvents;
    private long timestamp;
    
    public BatchWriteStrategy() {
        this.timestamp = System.currentTimeMillis();
    }
    
    /**
     * 获取Maxwell事件减少率
     * @return Maxwell事件减少率
     */
    public double getMaxwellEventReductionRate() {
        if (estimatedMaxwellEvents == 0) return 0.0;
        return (double) (batchSize - estimatedMaxwellEvents) / batchSize * 100;
    }
}

/**
 * Maxwell事件监控服务
 * @author 运维实战
 */
@Service
public class MaxwellEventMonitorService {
    
    private final AtomicLong totalBatchInserts = new AtomicLong(0);
    private final AtomicLong totalBatchUpdates = new AtomicLong(0);
    private final AtomicLong totalBatchDeletes = new AtomicLong(0);
    private final AtomicLong totalMaxwellEventsReduced = new AtomicLong(0);
    
    private long lastResetTime = System.currentTimeMillis();
    private final long resetInterval = 300000; // 5分钟重置一次
    
    private static final Logger logger = LoggerFactory.getLogger(MaxwellEventMonitorService.class);
    
    /**
     * 记录批量插入
     * @param tableName 表名
     * @param dataCount 数据数量
     * @param successCount 成功数量
     */
    public void recordBatchInsert(String tableName, int dataCount, int successCount) {
        totalBatchInserts.incrementAndGet();
        totalMaxwellEventsReduced.addAndGet(dataCount - 1); // 减少的事件数
        
        logger.debug("记录批量插入: 表名={}, 数据量={}, 成功数={}, 减少事件数={}", 
            tableName, dataCount, successCount, dataCount - 1);
    }
    
    /**
     * 记录批量更新
     * @param tableName 表名
     * @param dataCount 数据数量
     * @param successCount 成功数量
     */
    public void recordBatchUpdate(String tableName, int dataCount, int successCount) {
        totalBatchUpdates.incrementAndGet();
        totalMaxwellEventsReduced.addAndGet(dataCount - 1); // 减少的事件数
        
        logger.debug("记录批量更新: 表名={}, 数据量={}, 成功数={}, 减少事件数={}", 
            tableName, dataCount, successCount, dataCount - 1);
    }
    
    /**
     * 记录批量删除
     * @param tableName 表名
     * @param dataCount 数据数量
     * @param successCount 成功数量
     */
    public void recordBatchDelete(String tableName, int dataCount, int successCount) {
        totalBatchDeletes.incrementAndGet();
        totalMaxwellEventsReduced.addAndGet(dataCount - 1); // 减少的事件数
        
        logger.debug("记录批量删除: 表名={}, 数据量={}, 成功数={}, 减少事件数={}", 
            tableName, dataCount, successCount, dataCount - 1);
    }
    
    /**
     * 获取监控指标
     * @return 监控指标
     */
    public MaxwellEventMetrics getMetrics() {
        // 检查是否需要重置
        if (System.currentTimeMillis() - lastResetTime > resetInterval) {
            resetMetrics();
        }
        
        MaxwellEventMetrics metrics = new MaxwellEventMetrics();
        metrics.setTotalBatchInserts(totalBatchInserts.get());
        metrics.setTotalBatchUpdates(totalBatchUpdates.get());
        metrics.setTotalBatchDeletes(totalBatchDeletes.get());
        metrics.setTotalMaxwellEventsReduced(totalMaxwellEventsReduced.get());
        metrics.setTimestamp(System.currentTimeMillis());
        
        return metrics;
    }
    
    /**
     * 重置指标
     */
    private void resetMetrics() {
        totalBatchInserts.set(0);
        totalBatchUpdates.set(0);
        totalBatchDeletes.set(0);
        totalMaxwellEventsReduced.set(0);
        lastResetTime = System.currentTimeMillis();
        
        logger.info("Maxwell事件监控指标重置");
    }
    
    /**
     * 定期监控Maxwell事件状态
     */
    @Scheduled(fixedRate = 30000) // 每30秒监控一次
    public void monitorMaxwellEventStatus() {
        try {
            MaxwellEventMetrics metrics = getMetrics();
            
            logger.info("Maxwell事件监控: 批量插入={}, 批量更新={}, 批量删除={}, 总减少事件数={}, 事件减少率={}%", 
                metrics.getTotalBatchInserts(), metrics.getTotalBatchUpdates(), 
                metrics.getTotalBatchDeletes(), metrics.getTotalMaxwellEventsReduced(),
                String.format("%.2f", metrics.getEventReductionRate()));
            
            // 检查异常情况
            if (metrics.getEventReductionRate() < 50) {
                logger.warn("Maxwell事件减少率过低: {}%", String.format("%.2f", metrics.getEventReductionRate()));
            }
            
        } catch (Exception e) {
            logger.error("Maxwell事件状态监控失败", e);
        }
    }
}

/**
 * Maxwell事件指标类
 * @author 运维实战
 */
@Data
public class MaxwellEventMetrics {
    
    private long totalBatchInserts;
    private long totalBatchUpdates;
    private long totalBatchDeletes;
    private long totalMaxwellEventsReduced;
    private long timestamp;
    
    public MaxwellEventMetrics() {
        this.timestamp = System.currentTimeMillis();
    }
    
    /**
     * 获取总批量操作数
     * @return 总批量操作数
     */
    public long getTotalBatchOperations() {
        return totalBatchInserts + totalBatchUpdates + totalBatchDeletes;
    }
    
    /**
     * 获取事件减少率
     * @return 事件减少率
     */
    public double getEventReductionRate() {
        if (getTotalBatchOperations() == 0) return 0.0;
        return (double) totalMaxwellEventsReduced / getTotalBatchOperations() * 100;
    }
    
    /**
     * 是否健康
     * @return 是否健康
     */
    public boolean isHealthy() {
        return getEventReductionRate() > 50;
    }
}

/**
 * 批量写入策略服务
 * @author 运维实战
 */
@Service
public class BatchWriteStrategyService {
    
    @Autowired
    private MySQLBatchWriteProperties properties;
    
    private static final Logger logger = LoggerFactory.getLogger(BatchWriteStrategyService.class);
    
    /**
     * 获取批量写入策略
     * @param dataCount 数据数量
     * @return 批量写入策略
     */
    public BatchWriteStrategy getStrategy(int dataCount) {
        logger.info("获取批量写入策略，数据数量: {}", dataCount);
        
        BatchWriteStrategy strategy = new BatchWriteStrategy();
        strategy.setBatchSize(dataCount);
        strategy.setTimestamp(System.currentTimeMillis());
        
        // 根据数据量选择策略
        if (dataCount <= 100) {
            strategy.setStrategy("SMALL_BATCH");
            strategy.setDescription("小批量写入，Maxwell事件减少率较低");
            strategy.setEstimatedDuration(dataCount * 10);
            strategy.setEstimatedMaxwellEvents(dataCount);
        } else if (dataCount <= 500) {
            strategy.setStrategy("MEDIUM_BATCH");
            strategy.setDescription("中批量写入，Maxwell事件减少率中等");
            strategy.setEstimatedDuration(dataCount * 8);
            strategy.setEstimatedMaxwellEvents(1); // 批量操作只产生1个事件
        } else if (dataCount <= 1000) {
            strategy.setStrategy("LARGE_BATCH");
            strategy.setDescription("大批量写入，Maxwell事件减少率较高");
            strategy.setEstimatedDuration(dataCount * 6);
            strategy.setEstimatedMaxwellEvents(1); // 批量操作只产生1个事件
        } else {
            strategy.setStrategy("HUGE_BATCH");
            strategy.setDescription("超大批量写入，Maxwell事件减少率最高");
            strategy.setEstimatedDuration(dataCount * 5);
            strategy.setEstimatedMaxwellEvents(1); // 批量操作只产生1个事件
        }
        
        logger.info("批量写入策略确定: 策略={}, 数据量={}, 预计事件数={}, 事件减少率={}%", 
            strategy.getStrategy(), dataCount, strategy.getEstimatedMaxwellEvents(),
            String.format("%.2f", strategy.getMaxwellEventReductionRate()));
        
        return strategy;
    }
    
    /**
     * 计算最优批量大小
     * @param totalDataCount 总数据数量
     * @return 最优批量大小
     */
    public int calculateOptimalBatchSize(int totalDataCount) {
        if (totalDataCount <= 100) {
            return totalDataCount;
        } else if (totalDataCount <= 1000) {
            return Math.min(500, totalDataCount);
        } else if (totalDataCount <= 10000) {
            return Math.min(1000, totalDataCount);
        } else {
            return Math.min(properties.getMaxBatchSize(), totalDataCount);
        }
    }
    
    /**
     * 计算预计Maxwell事件数
     * @param dataCount 数据数量
     * @param batchSize 批量大小
     * @return 预计Maxwell事件数
     */
    public long calculateEstimatedMaxwellEvents(int dataCount, int batchSize) {
        if (batchSize >= dataCount) {
            return 1; // 单次批量操作
        } else {
            return (long) Math.ceil((double) dataCount / batchSize);
        }
    }
    
    /**
     * 计算Maxwell事件减少数
     * @param dataCount 数据数量
     * @param estimatedEvents 预计事件数
     * @return Maxwell事件减少数
     */
    public long calculateMaxwellEventReduction(int dataCount, long estimatedEvents) {
        return Math.max(0, dataCount - estimatedEvents);
    }
}

/**
 * 批量写MySQL REST控制器
 * @author 运维实战
 */
@RestController
@RequestMapping("/api/mysql/batch/write")
public class MySQLBatchWriteController {
    
    @Autowired
    private MySQLBatchWriteService mysqlBatchWriteService;
    
    @Autowired
    private MaxwellEventMonitorService maxwellEventMonitorService;
    
    @Autowired
    private BatchWriteStrategyService batchWriteStrategyService;
    
    private static final Logger logger = LoggerFactory.getLogger(MySQLBatchWriteController.class);
    
    /**
     * 批量插入减少Maxwell事件数
     * @param request 批量插入请求
     * @return 批量插入结果
     */
    @PostMapping("/insert")
    public ResponseEntity<MySQLBatchWriteResult> batchInsert(@RequestBody MySQLBatchWriteRequest request) {
        try {
            logger.info("接收到批量插入请求，表名: {}, 数据量: {}", request.getTableName(), request.getDataList().size());
            
            MySQLBatchWriteResult result = mysqlBatchWriteService.batchInsertWithMaxwellOptimization(
                request.getTableName(), request.getDataList());
            
            return ResponseEntity.ok(result);
            
        } catch (Exception e) {
            logger.error("批量插入失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    /**
     * 批量更新减少Maxwell事件数
     * @param request 批量更新请求
     * @return 批量更新结果
     */
    @PostMapping("/update")
    public ResponseEntity<MySQLBatchWriteResult> batchUpdate(@RequestBody MySQLBatchWriteRequest request) {
        try {
            logger.info("接收到批量更新请求，表名: {}, 数据量: {}", request.getTableName(), request.getDataList().size());
            
            MySQLBatchWriteResult result = mysqlBatchWriteService.batchUpdateWithMaxwellOptimization(
                request.getTableName(), request.getDataList());
            
            return ResponseEntity.ok(result);
            
        } catch (Exception e) {
            logger.error("批量更新失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    /**
     * 批量删除减少Maxwell事件数
     * @param request 批量删除请求
     * @return 批量删除结果
     */
    @PostMapping("/delete")
    public ResponseEntity<MySQLBatchWriteResult> batchDelete(@RequestBody MySQLBatchWriteRequest request) {
        try {
            logger.info("接收到批量删除请求，表名: {}, 数据量: {}", request.getTableName(), request.getDataList().size());
            
            MySQLBatchWriteResult result = mysqlBatchWriteService.batchDeleteWithMaxwellOptimization(
                request.getTableName(), request.getDataList());
            
            return ResponseEntity.ok(result);
            
        } catch (Exception e) {
            logger.error("批量删除失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    /**
     * 获取批量写入策略
     * @param dataCount 数据数量
     * @return 批量写入策略
     */
    @GetMapping("/strategy/{dataCount}")
    public ResponseEntity<BatchWriteStrategy> getBatchWriteStrategy(@PathVariable int dataCount) {
        try {
            BatchWriteStrategy strategy = batchWriteStrategyService.getStrategy(dataCount);
            return ResponseEntity.ok(strategy);
        } catch (Exception e) {
            logger.error("获取批量写入策略失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    /**
     * 获取Maxwell事件监控指标
     * @return Maxwell事件监控指标
     */
    @GetMapping("/maxwell/metrics")
    public ResponseEntity<MaxwellEventMetrics> getMaxwellEventMetrics() {
        try {
            MaxwellEventMetrics metrics = maxwellEventMonitorService.getMetrics();
            return ResponseEntity.ok(metrics);
        } catch (Exception e) {
            logger.error("获取Maxwell事件监控指标失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    /**
     * 计算最优批量大小
     * @param totalDataCount 总数据数量
     * @return 最优批量大小
     */
    @GetMapping("/optimal-batch-size/{totalDataCount}")
    public ResponseEntity<Map<String, Object>> getOptimalBatchSize(@PathVariable int totalDataCount) {
        try {
            int optimalBatchSize = batchWriteStrategyService.calculateOptimalBatchSize(totalDataCount);
            long estimatedEvents = batchWriteStrategyService.calculateEstimatedMaxwellEvents(totalDataCount, optimalBatchSize);
            long eventReduction = batchWriteStrategyService.calculateMaxwellEventReduction(totalDataCount, estimatedEvents);
            
            Map<String, Object> response = new HashMap<>();
            response.put("totalDataCount", totalDataCount);
            response.put("optimalBatchSize", optimalBatchSize);
            response.put("estimatedMaxwellEvents", estimatedEvents);
            response.put("maxwellEventReduction", eventReduction);
            response.put("eventReductionRate", (double) eventReduction / totalDataCount * 100);
            
            return ResponseEntity.ok(response);
            
        } catch (Exception e) {
            logger.error("计算最优批量大小失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
}

/**
 * MySQL批量写入请求类
 * @author 运维实战
 */
@Data
public class MySQLBatchWriteRequest {
    
    private String tableName;
    private List<Map<String, Object>> dataList;
    
    public MySQLBatchWriteRequest() {}
    
    public MySQLBatchWriteRequest(String tableName, List<Map<String, Object>> dataList) {
        this.tableName = tableName;
        this.dataList = dataList;
    }
}

/**
 * 批量写MySQL注解
 * @author 运维实战
 */
@Target({ElementType.METHOD, ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface MySQLBatchWrite {
    
    /**
     * 表名
     */
    String tableName() default "";
    
    /**
     * 最大批量大小
     */
    int maxBatchSize() default 1000;
    
    /**
     * 是否启用Maxwell事件优化
     */
    boolean enableMaxwellOptimization() default true;
    
    /**
     * 是否启用事务批量写入
     */
    boolean enableTransactionalBatch() default true;
    
    /**
     * 事务超时时间（毫秒）
     */
    long transactionTimeoutMs() default 30000;
    
    /**
     * 操作失败时的消息
     */
    String message() default "MySQL批量写操作失败，请稍后重试";
    
    /**
     * 操作失败时的HTTP状态码
     */
    int statusCode() default 500;
}

/**
 * 批量写MySQL AOP切面
 * @author 运维实战
 */
@Aspect
@Component
public class MySQLBatchWriteAspect {
    
    @Autowired
    private MySQLBatchWriteService mysqlBatchWriteService;
    
    @Autowired
    private MaxwellEventMonitorService maxwellEventMonitorService;
    
    private static final Logger logger = LoggerFactory.getLogger(MySQLBatchWriteAspect.class);
    
    /**
     * 批量写MySQL切点
     */
    @Pointcut("@annotation(mysqlBatchWrite)")
    public void mysqlBatchWritePointcut(MySQLBatchWrite mysqlBatchWrite) {}
    
    /**
     * 批量写MySQL环绕通知
     * @param joinPoint 连接点
     * @param mysqlBatchWrite 批量写注解
     * @return 执行结果
     * @throws Throwable 异常
     */
    @Around("mysqlBatchWritePointcut(mysqlBatchWrite)")
    public Object around(ProceedingJoinPoint joinPoint, MySQLBatchWrite mysqlBatchWrite) throws Throwable {
        String methodName = joinPoint.getSignature().getName();
        
        try {
            // 获取方法参数
            Object[] args = joinPoint.getArgs();
            
            // 查找数据参数
            List<Map<String, Object>> dataList = findDataList(args);
            String tableName = mysqlBatchWrite.tableName();
            
            if (dataList != null && !dataList.isEmpty()) {
                // 获取批量写入策略
                BatchWriteStrategy strategy = batchWriteStrategyService.getStrategy(dataList.size());
                
                logger.info("MySQL批量写操作开始: method={}, tableName={}, dataCount={}, strategy={}", 
                    methodName, tableName, dataList.size(), strategy.getStrategy());
                
                // 记录Maxwell事件优化指标
                maxwellEventMonitorService.recordBatchInsert(tableName, dataList.size(), dataList.size());
            }
            
            // 执行原方法
            return joinPoint.proceed();
            
        } catch (Exception e) {
            logger.error("MySQL批量写操作处理异常: method={}", methodName, e);
            throw new MySQLBatchWriteException(mysqlBatchWrite.message(), mysqlBatchWrite.statusCode());
        }
    }
    
    /**
     * 查找数据列表参数
     * @param args 方法参数
     * @return 数据列表
     */
    private List<Map<String, Object>> findDataList(Object[] args) {
        for (Object arg : args) {
            if (arg instanceof List) {
                @SuppressWarnings("unchecked")
                List<Map<String, Object>> list = (List<Map<String, Object>>) arg;
                return list;
            }
        }
        return null;
    }
    
    @Autowired
    private BatchWriteStrategyService batchWriteStrategyService;
}

/**
 * 批量写MySQL异常类
 * @author 运维实战
 */
public class MySQLBatchWriteException extends RuntimeException {
    
    private final int statusCode;
    
    public MySQLBatchWriteException(String message) {
        super(message);
        this.statusCode = 500;
    }
    
    public MySQLBatchWriteException(String message, int statusCode) {
        super(message);
        this.statusCode = statusCode;
    }
    
    public MySQLBatchWriteException(String message, Throwable cause) {
        super(message, cause);
        this.statusCode = 500;
    }
    
    public MySQLBatchWriteException(String message, Throwable cause, int statusCode) {
        super(message, cause);
        this.statusCode = statusCode;
    }
    
    public int getStatusCode() {
        return statusCode;
    }
}

/**
 * 批量写MySQL异常处理器
 * @author 运维实战
 */
@ControllerAdvice
public class MySQLBatchWriteExceptionHandler {
    
    private static final Logger logger = LoggerFactory.getLogger(MySQLBatchWriteExceptionHandler.class);
    
    /**
     * 处理批量写MySQL异常
     * @param e 异常
     * @return 错误响应
     */
    @ExceptionHandler(MySQLBatchWriteException.class)
    public ResponseEntity<Map<String, Object>> handleMySQLBatchWriteException(MySQLBatchWriteException e) {
        logger.warn("批量写MySQL异常: {}", e.getMessage());
        
        Map<String, Object> response = new HashMap<>();
        response.put("error", "MYSQL_BATCH_WRITE_FAILED");
        response.put("message", e.getMessage());
        response.put("timestamp", System.currentTimeMillis());
        
        return ResponseEntity.status(e.getStatusCode()).body(response);
    }
}

/**
 * 使用批量写MySQL注解的服务
 * @author 运维实战
 */
@Service
public class MySQLBatchWriteExampleService {
    
    private static final Logger logger = LoggerFactory.getLogger(MySQLBatchWriteExampleService.class);
    
    /**
     * 基础批量插入示例
     * @param dataList 数据列表
     * @return 处理结果
     */
    @MySQLBatchWrite(tableName = "user_data", maxBatchSize = 500, enableMaxwellOptimization = true,
        message = "基础批量插入：操作失败")
    public String basicBatchInsert(List<Map<String, Object>> dataList) {
        logger.info("执行基础批量插入示例，数据量: {}", dataList.size());
        
        // 模拟MySQL批量插入
        try {
            Thread.sleep(dataList.size() * 10);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        return "基础批量插入完成，数据量: " + dataList.size();
    }
    
    /**
     * 大批量插入示例
     * @param dataList 数据列表
     * @return 处理结果
     */
    @MySQLBatchWrite(tableName = "order_data", maxBatchSize = 2000, enableMaxwellOptimization = true,
        message = "大批量插入：操作失败")
    public String largeBatchInsert(List<Map<String, Object>> dataList) {
        logger.info("执行大批量插入示例，数据量: {}", dataList.size());
        
        // 模拟MySQL大批量插入
        try {
            Thread.sleep(dataList.size() * 5);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        return "大批量插入完成，数据量: " + dataList.size();
    }
    
    /**
     * 事务批量插入示例
     * @param dataList 数据列表
     * @return 处理结果
     */
    @MySQLBatchWrite(tableName = "product_data", maxBatchSize = 1000, enableTransactionalBatch = true,
        transactionTimeoutMs = 60000, message = "事务批量插入：操作失败")
    public String transactionalBatchInsert(List<Map<String, Object>> dataList) {
        logger.info("执行事务批量插入示例，数据量: {}", dataList.size());
        
        // 模拟MySQL事务批量插入
        try {
            Thread.sleep(dataList.size() * 8);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        return "事务批量插入完成，数据量: " + dataList.size();
    }
}

/**
 * 批量写MySQL测试控制器
 * @author 运维实战
 */
@RestController
@RequestMapping("/api/mysql/batch/write/test")
public class MySQLBatchWriteTestController {
    
    @Autowired
    private MySQLBatchWriteExampleService exampleService;
    
    @Autowired
    private MySQLBatchWriteService mysqlBatchWriteService;
    
    @Autowired
    private MaxwellEventMonitorService maxwellEventMonitorService;
    
    @Autowired
    private BatchWriteStrategyService batchWriteStrategyService;
    
    private static final Logger logger = LoggerFactory.getLogger(MySQLBatchWriteTestController.class);
    
    /**
     * 基础批量插入测试
     * @param dataCount 数据数量
     * @return 测试结果
     */
    @GetMapping("/basic-insert")
    public ResponseEntity<Map<String, String>> testBasicBatchInsert(@RequestParam int dataCount) {
        try {
            // 生成测试数据
            List<Map<String, Object>> dataList = generateTestData(dataCount);
            
            String result = exampleService.basicBatchInsert(dataList);
            
            Map<String, String> response = new HashMap<>();
            response.put("status", "SUCCESS");
            response.put("result", result);
            response.put("timestamp", String.valueOf(System.currentTimeMillis()));
            
            return ResponseEntity.ok(response);
            
        } catch (MySQLBatchWriteException e) {
            logger.warn("基础批量插入测试失败: {}", e.getMessage());
            return ResponseEntity.status(e.getStatusCode()).build();
        } catch (Exception e) {
            logger.error("基础批量插入测试失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    /**
     * 大批量插入测试
     * @param dataCount 数据数量
     * @return 测试结果
     */
    @GetMapping("/large-insert")
    public ResponseEntity<Map<String, String>> testLargeBatchInsert(@RequestParam int dataCount) {
        try {
            // 生成测试数据
            List<Map<String, Object>> dataList = generateTestData(dataCount);
            
            String result = exampleService.largeBatchInsert(dataList);
            
            Map<String, String> response = new HashMap<>();
            response.put("status", "SUCCESS");
            response.put("result", result);
            response.put("timestamp", String.valueOf(System.currentTimeMillis()));
            
            return ResponseEntity.ok(response);
            
        } catch (MySQLBatchWriteException e) {
            logger.warn("大批量插入测试失败: {}", e.getMessage());
            return ResponseEntity.status(e.getStatusCode()).build();
        } catch (Exception e) {
            logger.error("大批量插入测试失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    /**
     * 事务批量插入测试
     * @param dataCount 数据数量
     * @return 测试结果
     */
    @GetMapping("/transactional-insert")
    public ResponseEntity<Map<String, String>> testTransactionalBatchInsert(@RequestParam int dataCount) {
        try {
            // 生成测试数据
            List<Map<String, Object>> dataList = generateTestData(dataCount);
            
            String result = exampleService.transactionalBatchInsert(dataList);
            
            Map<String, String> response = new HashMap<>();
            response.put("status", "SUCCESS");
            response.put("result", result);
            response.put("timestamp", String.valueOf(System.currentTimeMillis()));
            
            return ResponseEntity.ok(response);
            
        } catch (MySQLBatchWriteException e) {
            logger.warn("事务批量插入测试失败: {}", e.getMessage());
            return ResponseEntity.status(e.getStatusCode()).build();
        } catch (Exception e) {
            logger.error("事务批量插入测试失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    /**
     * 获取批量写入策略
     * @param dataCount 数据数量
     * @return 批量写入策略
     */
    @GetMapping("/strategy/{dataCount}")
    public ResponseEntity<BatchWriteStrategy> getBatchWriteStrategy(@PathVariable int dataCount) {
        try {
            BatchWriteStrategy strategy = batchWriteStrategyService.getStrategy(dataCount);
            return ResponseEntity.ok(strategy);
        } catch (Exception e) {
            logger.error("获取批量写入策略失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    /**
     * 获取Maxwell事件监控指标
     * @return Maxwell事件监控指标
     */
    @GetMapping("/maxwell/metrics")
    public ResponseEntity<MaxwellEventMetrics> getMaxwellEventMetrics() {
        try {
            MaxwellEventMetrics metrics = maxwellEventMonitorService.getMetrics();
            return ResponseEntity.ok(metrics);
        } catch (Exception e) {
            logger.error("获取Maxwell事件监控指标失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    /**
     * 生成测试数据
     * @param count 数据数量
     * @return 测试数据
     */
    private List<Map<String, Object>> generateTestData(int count) {
        List<Map<String, Object>> dataList = new ArrayList<>();
        
        for (int i = 0; i < count; i++) {
            Map<String, Object> data = new HashMap<>();
            data.put("id", i);
            data.put("name", "test_name_" + i);
            data.put("email", "test_email_" + i + "@example.com");
            data.put("age", 20 + (i % 50));
            data.put("created_at", new Timestamp(System.currentTimeMillis()));
            dataList.add(data);
        }
        
        return dataList;
    }
}