👋 大家好，我是 阿问学长！专注于分享优质开源项目解析、毕业设计项目指导支持、幼小初高的教辅资料推荐等，欢迎关注交流！🚀

数据库连接与事务管理

前言

数据库连接池和事务管理是企业级应用的核心组件，直接影响系统的性能、稳定性和数据一致性。RuoYi-Vue-Plus框架基于HikariCP连接池和Spring事务管理，提供了高性能的数据库连接管理和灵活的事务控制机制。本文将详细介绍HikariCP连接池的配置优化、Spring事务管理的高级用法、分布式事务处理以及连接池监控等内容。

HikariCP连接池优化

连接池基础配置

/**
 * HikariCP连接池配置
 */
@Configuration
public class HikariCPConfig {
    
    /**
     * 主数据源连接池配置
     */
    @Bean
    @Primary
    @ConfigurationProperties("spring.datasource.hikari")
    public HikariConfig primaryHikariConfig() {
        HikariConfig config = new HikariConfig();
        
        // 基础连接配置
        config.setJdbcUrl("jdbc:mysql://localhost:3306/ry-vue-plus");
        config.setUsername("root");
        config.setPassword("password");
        config.setDriverClassName("com.mysql.cj.jdbc.Driver");
        
        // 连接池大小配置
        config.setMaximumPoolSize(20);           // 最大连接数
        config.setMinimumIdle(5);                // 最小空闲连接数
        
        // 连接超时配置
        config.setConnectionTimeout(30000);      // 连接超时时间(ms)
        config.setIdleTimeout(600000);          // 空闲连接超时时间(ms)
        config.setMaxLifetime(1800000);         // 连接最大生存时间(ms)
        config.setValidationTimeout(5000);      // 连接验证超时时间(ms)
        
        // 连接测试配置
        config.setConnectionTestQuery("SELECT 1");
        config.setTestWhileIdle(true);
        
        // 连接池名称
        config.setPoolName("HikariCP-Primary");
        
        // 性能优化配置
        config.setLeakDetectionThreshold(60000); // 连接泄漏检测阈值(ms)
        config.setRegisterMbeans(true);         // 注册JMX MBeans
        
        // 连接属性配置
        config.addDataSourceProperty("cachePrepStmts", "true");
        config.addDataSourceProperty("prepStmtCacheSize", "250");
        config.addDataSourceProperty("prepStmtCacheSqlLimit", "2048");
        config.addDataSourceProperty("useServerPrepStmts", "true");
        config.addDataSourceProperty("useLocalSessionState", "true");
        config.addDataSourceProperty("rewriteBatchedStatements", "true");
        config.addDataSourceProperty("cacheResultSetMetadata", "true");
        config.addDataSourceProperty("cacheServerConfiguration", "true");
        config.addDataSourceProperty("elideSetAutoCommits", "true");
        config.addDataSourceProperty("maintainTimeStats", "false");
        
        return config;
    }
    
    /**
     * 从数据源连接池配置（读优化）
     */
    @Bean
    @ConfigurationProperties("spring.datasource.slave.hikari")
    public HikariConfig slaveHikariConfig() {
        HikariConfig config = new HikariConfig();
        
        // 基础连接配置
        config.setJdbcUrl("jdbc:mysql://localhost:3307/ry-vue-plus");
        config.setUsername("root");
        config.setPassword("password");
        config.setDriverClassName("com.mysql.cj.jdbc.Driver");
        
        // 读库连接池配置（可以设置更多连接）
        config.setMaximumPoolSize(30);
        config.setMinimumIdle(10);
        
        // 读库超时配置（可以设置更短的超时时间）
        config.setConnectionTimeout(20000);
        config.setIdleTimeout(300000);
        config.setMaxLifetime(1200000);
        config.setValidationTimeout(3000);
        
        config.setConnectionTestQuery("SELECT 1");
        config.setPoolName("HikariCP-Slave");
        
        // 读库优化配置
        config.addDataSourceProperty("cachePrepStmts", "true");
        config.addDataSourceProperty("prepStmtCacheSize", "500");
        config.addDataSourceProperty("prepStmtCacheSqlLimit", "4096");
        config.addDataSourceProperty("useServerPrepStmts", "true");
        config.addDataSourceProperty("cacheResultSetMetadata", "true");
        config.addDataSourceProperty("defaultFetchSize", "1000");
        
        return config;
    }
    
    /**
     * 连接池健康检查
     */
    @Bean
    public HikariPoolHealthIndicator hikariPoolHealthIndicator() {
        return new HikariPoolHealthIndicator();
    }
}

连接池性能调优

/**
 * 连接池性能调优服务
 */
@Service
public class ConnectionPoolTuningService {
    
    private static final Logger log = LoggerFactory.getLogger(ConnectionPoolTuningService.class);
    
    @Autowired
    private HikariDataSource primaryDataSource;
    
    @Autowired
    private HikariDataSource slaveDataSource;
    
    /**
     * 获取连接池状态
     */
    public ConnectionPoolStatus getPoolStatus(String poolName) {
        HikariDataSource dataSource = getDataSourceByName(poolName);
        if (dataSource == null) {
            return null;
        }
        
        HikariPoolMXBean poolMXBean = dataSource.getHikariPoolMXBean();
        
        ConnectionPoolStatus status = new ConnectionPoolStatus();
        status.setPoolName(poolName);
        status.setActiveConnections(poolMXBean.getActiveConnections());
        status.setIdleConnections(poolMXBean.getIdleConnections());
        status.setTotalConnections(poolMXBean.getTotalConnections());
        status.setThreadsAwaitingConnection(poolMXBean.getThreadsAwaitingConnection());
        status.setMaximumPoolSize(dataSource.getMaximumPoolSize());
        status.setMinimumIdle(dataSource.getMinimumIdle());
        
        // 计算使用率
        double usageRate = (double) status.getActiveConnections() / status.getMaximumPoolSize();
        status.setUsageRate(usageRate);
        
        return status;
    }
    
    /**
     * 动态调整连接池大小
     */
    public void adjustPoolSize(String poolName, int newMaxSize, int newMinIdle) {
        HikariDataSource dataSource = getDataSourceByName(poolName);
        if (dataSource == null) {
            log.warn("数据源不存在: {}", poolName);
            return;
        }
        
        int oldMaxSize = dataSource.getMaximumPoolSize();
        int oldMinIdle = dataSource.getMinimumIdle();
        
        // 调整连接池大小
        dataSource.setMaximumPoolSize(newMaxSize);
        dataSource.setMinimumIdle(newMinIdle);
        
        log.info("连接池大小已调整: pool={}, maxSize: {} -> {}, minIdle: {} -> {}", 
                poolName, oldMaxSize, newMaxSize, oldMinIdle, newMinIdle);
    }
    
    /**
     * 连接池预热
     */
    public void warmUpPool(String poolName) {
        HikariDataSource dataSource = getDataSourceByName(poolName);
        if (dataSource == null) {
            return;
        }
        
        int minIdle = dataSource.getMinimumIdle();
        List<Connection> connections = new ArrayList<>();
        
        try {
            // 创建最小空闲连接数的连接
            for (int i = 0; i < minIdle; i++) {
                Connection connection = dataSource.getConnection();
                connections.add(connection);
            }
            
            log.info("连接池预热完成: pool={}, connections={}", poolName, minIdle);
            
        } catch (SQLException e) {
            log.error("连接池预热失败: pool={}", poolName, e);
        } finally {
            // 释放连接
            for (Connection connection : connections) {
                try {
                    connection.close();
                } catch (SQLException e) {
                    log.warn("关闭连接失败", e);
                }
            }
        }
    }
    
    /**
     * 连接池性能监控
     */
    @Scheduled(fixedRate = 60000) // 每分钟监控一次
    public void monitorConnectionPools() {
        monitorPool("primary", primaryDataSource);
        monitorPool("slave", slaveDataSource);
    }
    
    /**
     * 监控单个连接池
     */
    private void monitorPool(String poolName, HikariDataSource dataSource) {
        try {
            ConnectionPoolStatus status = getPoolStatus(poolName);
            
            // 记录监控指标
            recordPoolMetrics(status);
            
            // 检查告警条件
            checkPoolAlerts(status);
            
        } catch (Exception e) {
            log.error("连接池监控异常: pool={}", poolName, e);
        }
    }
    
    /**
     * 记录连接池指标
     */
    private void recordPoolMetrics(ConnectionPoolStatus status) {
        String poolName = status.getPoolName();
        
        // 使用Micrometer记录指标
        Metrics.gauge("hikari.connections.active", Tags.of("pool", poolName), status.getActiveConnections());
        Metrics.gauge("hikari.connections.idle", Tags.of("pool", poolName), status.getIdleConnections());
        Metrics.gauge("hikari.connections.total", Tags.of("pool", poolName), status.getTotalConnections());
        Metrics.gauge("hikari.connections.pending", Tags.of("pool", poolName), status.getThreadsAwaitingConnection());
        Metrics.gauge("hikari.connections.usage", Tags.of("pool", poolName), status.getUsageRate());
    }
    
    /**
     * 检查连接池告警
     */
    private void checkPoolAlerts(ConnectionPoolStatus status) {
        String poolName = status.getPoolName();
        
        // 使用率告警
        if (status.getUsageRate() > 0.9) {
            log.warn("连接池使用率过高: pool={}, usage={}%", poolName, status.getUsageRate() * 100);
            sendPoolAlert(poolName, "连接池使用率过高", status);
        }
        
        // 等待连接告警
        if (status.getThreadsAwaitingConnection() > 5) {
            log.warn("连接池等待连接数过多: pool={}, waiting={}", poolName, status.getThreadsAwaitingConnection());
            sendPoolAlert(poolName, "等待连接数过多", status);
        }
        
        // 空闲连接不足告警
        if (status.getIdleConnections() == 0 && status.getActiveConnections() > 0) {
            log.warn("连接池无空闲连接: pool={}", poolName);
            sendPoolAlert(poolName, "无空闲连接", status);
        }
    }
    
    /**
     * 发送连接池告警
     */
    private void sendPoolAlert(String poolName, String alertType, ConnectionPoolStatus status) {
        ConnectionPoolAlert alert = new ConnectionPoolAlert();
        alert.setPoolName(poolName);
        alert.setAlertType(alertType);
        alert.setStatus(status);
        alert.setTimestamp(new Date());
        
        // 异步发送告警
        AsyncManager.me().execute(() -> {
            alertService.sendConnectionPoolAlert(alert);
        });
    }
    
    /**
     * 根据名称获取数据源
     */
    private HikariDataSource getDataSourceByName(String poolName) {
        switch (poolName) {
            case "primary":
                return primaryDataSource;
            case "slave":
                return slaveDataSource;
            default:
                return null;
        }
    }
}

/**
 * 连接池状态实体
 */
@Data
public class ConnectionPoolStatus {
    private String poolName;
    private int activeConnections;
    private int idleConnections;
    private int totalConnections;
    private int threadsAwaitingConnection;
    private int maximumPoolSize;
    private int minimumIdle;
    private double usageRate;
}

Spring事务管理

声明式事务配置

/**
 * 事务管理配置
 */
@Configuration
@EnableTransactionManagement
public class TransactionConfig {
    
    /**
     * 主数据源事务管理器
     */
    @Bean
    @Primary
    public PlatformTransactionManager primaryTransactionManager(
            @Qualifier("primaryDataSource") DataSource dataSource) {
        DataSourceTransactionManager transactionManager = new DataSourceTransactionManager();
        transactionManager.setDataSource(dataSource);
        transactionManager.setRollbackOnCommitFailure(true);
        return transactionManager;
    }
    
    /**
     * 从数据源事务管理器
     */
    @Bean
    public PlatformTransactionManager slaveTransactionManager(
            @Qualifier("slaveDataSource") DataSource dataSource) {
        DataSourceTransactionManager transactionManager = new DataSourceTransactionManager();
        transactionManager.setDataSource(dataSource);
        transactionManager.setRollbackOnCommitFailure(true);
        return transactionManager;
    }
    
    /**
     * 事务模板（编程式事务）
     */
    @Bean
    public TransactionTemplate transactionTemplate(
            @Qualifier("primaryTransactionManager") PlatformTransactionManager transactionManager) {
        TransactionTemplate template = new TransactionTemplate(transactionManager);
        template.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
        template.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);
        template.setTimeout(30); // 30秒超时
        return template;
    }
    
    /**
     * 只读事务模板
     */
    @Bean
    public TransactionTemplate readOnlyTransactionTemplate(
            @Qualifier("slaveTransactionManager") PlatformTransactionManager transactionManager) {
        TransactionTemplate template = new TransactionTemplate(transactionManager);
        template.setReadOnly(true);
        template.setPropagationBehavior(TransactionDefinition.PROPAGATION_SUPPORTS);
        template.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);
        return template;
    }
}

事务传播行为应用

/**
 * 事务传播行为示例服务
 */
@Service
public class TransactionPropagationService {
    
    @Autowired
    private SysUserMapper userMapper;
    
    @Autowired
    private SysLogMapper logMapper;
    
    @Autowired
    private TransactionTemplate transactionTemplate;
    
    /**
     * REQUIRED - 默认传播行为
     * 如果当前存在事务，则加入该事务；如果当前没有事务，则创建一个新的事务
     */
    @Transactional(propagation = Propagation.REQUIRED)
    public void requiredExample(SysUserBo user) {
        // 主业务操作
        userMapper.insert(BeanUtil.toBean(user, SysUser.class));
        
        // 调用其他服务方法（会加入当前事务）
        logUserOperation(user.getUserId(), "CREATE_USER");
        
        // 如果这里抛出异常，整个事务都会回滚
        if (user.getUserName().contains("error")) {
            throw new ServiceException("用户名包含错误标识");
        }
    }
    
    /**
     * REQUIRES_NEW - 创建新事务
     * 总是创建一个新的事务，如果当前存在事务，则把当前事务挂起
     */
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void requiresNewExample(Long userId, String operation) {
        // 这个方法总是在新事务中执行
        SysLog log = new SysLog();
        log.setUserId(userId);
        log.setOperation(operation);
        log.setCreateTime(new Date());
        
        logMapper.insert(log);
        
        // 即使调用方的事务回滚，这个日志记录也会被提交
    }
    
    /**
     * NESTED - 嵌套事务
     * 如果当前存在事务，则在嵌套事务内执行；如果当前没有事务，则创建一个新的事务
     */
    @Transactional(propagation = Propagation.NESTED)
    public void nestedExample(SysUserBo user) {
        try {
            // 嵌套事务中的操作
            userMapper.insert(BeanUtil.toBean(user, SysUser.class));
            
            // 调用可能失败的操作
            riskyOperation(user);
            
        } catch (Exception e) {
            // 嵌套事务可以单独回滚，不影响外层事务
            log.warn("嵌套事务操作失败，但不影响主事务", e);
            throw e;
        }
    }
    
    /**
     * SUPPORTS - 支持事务
     * 如果当前存在事务，则加入该事务；如果当前没有事务，则以非事务方式执行
     */
    @Transactional(propagation = Propagation.SUPPORTS, readOnly = true)
    public SysUserVo supportsExample(Long userId) {
        // 这个方法支持事务，但不强制要求
        SysUser user = userMapper.selectById(userId);
        return BeanUtil.toBean(user, SysUserVo.class);
    }
    
    /**
     * NOT_SUPPORTED - 不支持事务
     * 以非事务方式执行，如果当前存在事务，则把当前事务挂起
     */
    @Transactional(propagation = Propagation.NOT_SUPPORTED)
    public void notSupportedExample(String message) {
        // 这个方法总是以非事务方式执行
        // 通常用于日志记录、缓存操作等不需要事务的操作
        log.info("非事务操作: {}", message);
    }
    
    /**
     * MANDATORY - 强制事务
     * 必须在事务中执行，如果当前没有事务，则抛出异常
     */
    @Transactional(propagation = Propagation.MANDATORY)
    public void mandatoryExample(SysUserBo user) {
        // 这个方法必须在事务中调用
        userMapper.updateById(BeanUtil.toBean(user, SysUser.class));
    }
    
    /**
     * NEVER - 禁止事务
     * 以非事务方式执行，如果当前存在事务，则抛出异常
     */
    @Transactional(propagation = Propagation.NEVER)
    public void neverExample(String cacheKey, Object value) {
        // 这个方法禁止在事务中调用
        // 通常用于缓存操作
        redisTemplate.opsForValue().set(cacheKey, value);
    }
    
    /**
     * 复杂事务场景示例
     */
    @Transactional(rollbackFor = Exception.class)
    public void complexTransactionExample(SysUserBo user) {
        // 1. 主业务操作（参与当前事务）
        userMapper.insert(BeanUtil.toBean(user, SysUser.class));
        
        // 2. 记录操作日志（新事务，即使主事务回滚也会提交）
        requiresNewExample(user.getUserId(), "CREATE_USER");
        
        // 3. 尝试发送通知（嵌套事务，失败不影响主事务）
        try {
            sendNotification(user);
        } catch (Exception e) {
            log.warn("发送通知失败，但不影响用户创建", e);
        }
        
        // 4. 更新缓存（非事务操作）
        updateUserCache(user);
    }
    
    @Transactional(propagation = Propagation.NESTED)
    private void sendNotification(SysUserBo user) {
        // 发送通知的逻辑
        notificationService.sendWelcomeEmail(user.getEmail());
    }
    
    @Transactional(propagation = Propagation.NOT_SUPPORTED)
    private void updateUserCache(SysUserBo user) {
        // 更新缓存
        String cacheKey = "user:" + user.getUserId();
        redisTemplate.opsForValue().set(cacheKey, user);
    }
    
    private void logUserOperation(Long userId, String operation) {
        SysLog log = new SysLog();
        log.setUserId(userId);
        log.setOperation(operation);
        log.setCreateTime(new Date());
        logMapper.insert(log);
    }
    
    private void riskyOperation(SysUserBo user) {
        // 模拟可能失败的操作
        if (user.getUserName().contains("risky")) {
            throw new ServiceException("风险操作失败");
        }
    }
}

编程式事务管理

/**
 * 编程式事务管理服务
 */
@Service
public class ProgrammaticTransactionService {
    
    @Autowired
    private TransactionTemplate transactionTemplate;
    
    @Autowired
    private TransactionTemplate readOnlyTransactionTemplate;
    
    @Autowired
    private PlatformTransactionManager transactionManager;
    
    @Autowired
    private SysUserMapper userMapper;
    
    /**
     * 使用TransactionTemplate的编程式事务
     */
    public Boolean createUserWithTemplate(SysUserBo user) {
        return transactionTemplate.execute(status -> {
            try {
                // 业务逻辑
                SysUser sysUser = BeanUtil.toBean(user, SysUser.class);
                int result = userMapper.insert(sysUser);
                
                // 模拟业务校验
                if (user.getUserName().contains("invalid")) {
                    // 手动回滚
                    status.setRollbackOnly();
                    return false;
                }
                
                // 其他业务操作
                processUserCreation(sysUser);
                
                return result > 0;
                
            } catch (Exception e) {
                log.error("创建用户失败", e);
                // 异常会自动回滚
                throw new ServiceException("创建用户失败: " + e.getMessage());
            }
        });
    }
    
    /**
     * 使用TransactionManager的编程式事务
     */
    public Boolean updateUserWithManager(SysUserBo user) {
        DefaultTransactionDefinition def = new DefaultTransactionDefinition();
        def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
        def.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);
        def.setTimeout(30);
        
        TransactionStatus status = transactionManager.getTransaction(def);
        
        try {
            // 业务逻辑
            SysUser sysUser = BeanUtil.toBean(user, SysUser.class);
            int result = userMapper.updateById(sysUser);
            
            // 业务校验
            if (result == 0) {
                throw new ServiceException("用户不存在或更新失败");
            }
            
            // 其他业务操作
            processUserUpdate(sysUser);
            
            // 提交事务
            transactionManager.commit(status);
            return true;
            
        } catch (Exception e) {
            // 回滚事务
            transactionManager.rollback(status);
            log.error("更新用户失败", e);
            throw new ServiceException("更新用户失败: " + e.getMessage());
        }
    }
    
    /**
     * 只读事务示例
     */
    public List<SysUserVo> getUserListWithReadOnlyTransaction(SysUserBo queryUser) {
        return readOnlyTransactionTemplate.execute(status -> {
            // 只读事务中的查询操作
            LambdaQueryWrapper<SysUser> wrapper = new LambdaQueryWrapper<>();
            wrapper.like(StringUtils.isNotBlank(queryUser.getUserName()), 
                        SysUser::getUserName, queryUser.getUserName());
            wrapper.eq(StringUtils.isNotBlank(queryUser.getStatus()), 
                      SysUser::getStatus, queryUser.getStatus());
            
            List<SysUser> users = userMapper.selectList(wrapper);
            return BeanUtil.copyToList(users, SysUserVo.class);
        });
    }
    
    /**
     * 复杂的编程式事务场景
     */
    public void complexProgrammaticTransaction(List<SysUserBo> users) {
        // 外层事务
        transactionTemplate.execute(outerStatus -> {
            
            for (SysUserBo user : users) {
                // 为每个用户创建独立的嵌套事务
                DefaultTransactionDefinition nestedDef = new DefaultTransactionDefinition();
                nestedDef.setPropagationBehavior(TransactionDefinition.PROPAGATION_NESTED);
                
                TransactionStatus nestedStatus = transactionManager.getTransaction(nestedDef);
                
                try {
                    // 处理单个用户
                    processUser(user);
                    
                    // 提交嵌套事务
                    transactionManager.commit(nestedStatus);
                    
                } catch (Exception e) {
                    // 回滚嵌套事务，但不影响外层事务
                    transactionManager.rollback(nestedStatus);
                    log.warn("处理用户失败，跳过: userId={}", user.getUserId(), e);
                }
            }
            
            return null;
        });
    }
    
    /**
     * 事务同步回调示例
     */
    public void transactionWithSynchronization(SysUserBo user) {
        transactionTemplate.execute(status -> {
            
            // 注册事务同步回调
            if (TransactionSynchronizationManager.isSynchronizationActive()) {
                TransactionSynchronizationManager.registerSynchronization(
                    new TransactionSynchronization() {
                        
                        @Override
                        public void beforeCommit(boolean readOnly) {
                            log.info("事务即将提交: readOnly={}", readOnly);
                        }
                        
                        @Override
                        public void afterCommit() {
                            log.info("事务提交成功");
                            // 事务提交后的操作，如清除缓存、发送消息等
                            clearUserCache(user.getUserId());
                            sendUserCreatedEvent(user);
                        }
                        
                        @Override
                        public void afterCompletion(int status) {
                            if (status == STATUS_COMMITTED) {
                                log.info("事务完成: 已提交");
                            } else if (status == STATUS_ROLLED_BACK) {
                                log.info("事务完成: 已回滚");
                            }
                        }
                    }
                );
            }
            
            // 业务逻辑
            SysUser sysUser = BeanUtil.toBean(user, SysUser.class);
            userMapper.insert(sysUser);
            
            return null;
        });
    }
    
    private void processUserCreation(SysUser user) {
        // 用户创建后的处理逻辑
    }
    
    private void processUserUpdate(SysUser user) {
        // 用户更新后的处理逻辑
    }
    
    private void processUser(SysUserBo user) {
        // 处理单个用户的逻辑
        SysUser sysUser = BeanUtil.toBean(user, SysUser.class);
        userMapper.insert(sysUser);
    }
    
    private void clearUserCache(Long userId) {
        // 清除用户缓存
        String cacheKey = "user:" + userId;
        redisTemplate.delete(cacheKey);
    }
    
    private void sendUserCreatedEvent(SysUserBo user) {
        // 发送用户创建事件
        eventPublisher.publishEvent(new UserCreatedEvent(user));
    }
}

分布式事务处理

Seata分布式事务配置

/**
 * Seata分布式事务配置
 */
@Configuration
public class SeataConfig {
    
    /**
     * 数据源代理（AT模式）
     */
    @Bean
    @Primary
    public DataSource dataSourceProxy(@Qualifier("primaryDataSource") DataSource dataSource) {
        return new DataSourceProxy(dataSource);
    }
    
    /**
     * 全局事务扫描器
     */
    @Bean
    public GlobalTransactionScanner globalTransactionScanner() {
        return new GlobalTransactionScanner("ruoyi-application", "default");
    }
    
    /**
     * Seata事务管理器
     */
    @Bean
    public SeataTransactionManager seataTransactionManager() {
        return new SeataTransactionManager();
    }
}

/**
 * 分布式事务服务示例
 */
@Service
public class DistributedTransactionService {
    
    @Autowired
    private SysUserMapper userMapper;
    
    @Autowired
    private OrderService orderService;
    
    @Autowired
    private InventoryService inventoryService;
    
    @Autowired
    private PaymentService paymentService;
    
    /**
     * 分布式事务示例 - 订单创建
     */
    @GlobalTransactional(name = "create-order", rollbackFor = Exception.class)
    public OrderResult createOrder(OrderRequest request) {
        try {
            // 1. 创建订单（本地事务）
            Order order = orderService.createOrder(request);
            log.info("订单创建成功: orderId={}", order.getOrderId());
            
            // 2. 扣减库存（远程服务调用）
            InventoryResult inventoryResult = inventoryService.reduceInventory(
                request.getProductId(), request.getQuantity());
            
            if (!inventoryResult.isSuccess()) {
                throw new ServiceException("库存扣减失败: " + inventoryResult.getMessage());
            }
            log.info("库存扣减成功: productId={}, quantity={}", 
                    request.getProductId(), request.getQuantity());
            
            // 3. 创建支付订单（远程服务调用）
            PaymentResult paymentResult = paymentService.createPayment(
                order.getOrderId(), request.getAmount());
            
            if (!paymentResult.isSuccess()) {
                throw new ServiceException("支付订单创建失败: " + paymentResult.getMessage());
            }
            log.info("支付订单创建成功: orderId={}, amount={}", 
                    order.getOrderId(), request.getAmount());
            
            // 4. 更新用户积分（本地事务）
            updateUserPoints(request.getUserId(), request.getAmount());
            
            return OrderResult.success(order);
            
        } catch (Exception e) {
            log.error("创建订单失败，分布式事务将回滚", e);
            throw e;
        }
    }
    
    /**
     * TCC模式分布式事务示例
     */
    @GlobalTransactional(name = "tcc-transfer", rollbackFor = Exception.class)
    public TransferResult transfer(TransferRequest request) {
        try {
            // 1. Try阶段 - 冻结转出账户资金
            boolean freezeResult = accountService.freezeAmount(
                request.getFromAccount(), request.getAmount());
            
            if (!freezeResult) {
                throw new ServiceException("资金冻结失败");
            }
            
            // 2. Try阶段 - 预留转入账户额度
            boolean reserveResult = accountService.reserveAmount(
                request.getToAccount(), request.getAmount());
            
            if (!reserveResult) {
                throw new ServiceException("额度预留失败");
            }
            
            // 3. 记录转账流水
            TransferRecord record = transferService.createTransferRecord(request);
            
            return TransferResult.success(record);
            
        } catch (Exception e) {
            log.error("转账失败，TCC事务将回滚", e);
            throw e;
        }
    }
    
    /**
     * Saga模式分布式事务示例
     */
    @SagaOrchestrationStart
    public void sagaOrderProcess(OrderRequest request) {
        // Saga编排器会按顺序执行以下步骤
        // 每个步骤都有对应的补偿操作
        
        // 步骤1: 创建订单
        sagaManager.choreography()
            .step("createOrder")
            .localTransaction(this::createOrderStep)
            .compensationMethod(this::cancelOrderStep)
            
            // 步骤2: 扣减库存
            .step("reduceInventory")
            .remoteTransaction("inventory-service", "/inventory/reduce")
            .compensationMethod("inventory-service", "/inventory/restore")
            
            // 步骤3: 创建支付
            .step("createPayment")
            .remoteTransaction("payment-service", "/payment/create")
            .compensationMethod("payment-service", "/payment/cancel")
            
            // 步骤4: 发送通知
            .step("sendNotification")
            .localTransaction(this::sendNotificationStep)
            .compensationMethod(this::cancelNotificationStep)
            
            .execute(request);
    }
    
    private void updateUserPoints(Long userId, BigDecimal amount) {
        // 更新用户积分的逻辑
        int points = amount.multiply(new BigDecimal("0.01")).intValue();
        userMapper.updateUserPoints(userId, points);
    }
    
    private void createOrderStep(OrderRequest request) {
        // 创建订单步骤
    }
    
    private void cancelOrderStep(OrderRequest request) {
        // 取消订单补偿步骤
    }
    
    private void sendNotificationStep(OrderRequest request) {
        // 发送通知步骤
    }
    
    private void cancelNotificationStep(OrderRequest request) {
        // 取消通知补偿步骤
    }
}

分布式事务监控

/**
 * 分布式事务监控服务
 */
@Service
public class DistributedTransactionMonitorService {
    
    @Autowired
    private GlobalTransactionScanner globalTransactionScanner;
    
    /**
     * 获取全局事务状态
     */
    public GlobalTransactionStatus getGlobalTransactionStatus(String xid) {
        try {
            // 查询全局事务状态
            GlobalStatus status = TransactionManagerHolder.get().getStatus(xid);
            
            GlobalTransactionStatus transactionStatus = new GlobalTransactionStatus();
            transactionStatus.setXid(xid);
            transactionStatus.setStatus(status.name());
            transactionStatus.setQueryTime(new Date());
            
            return transactionStatus;
            
        } catch (Exception e) {
            log.error("查询全局事务状态失败: xid={}", xid, e);
            return null;
        }
    }
    
    /**
     * 监控分布式事务性能
     */
    @EventListener
    public void handleGlobalTransactionEvent(GlobalTransactionEvent event) {
        String xid = event.getId();
        GlobalStatus status = event.getStatus();
        
        log.info("全局事务状态变更: xid={}, status={}", xid, status);
        
        // 记录事务性能指标
        recordTransactionMetrics(event);
        
        // 检查事务异常
        if (status == GlobalStatus.TimeoutRollbacked || 
            status == GlobalStatus.TimeoutRollbackFailed) {
            handleTransactionTimeout(event);
        }
    }
    
    /**
     * 记录事务指标
     */
    private void recordTransactionMetrics(GlobalTransactionEvent event) {
        String transactionName = event.getName();
        GlobalStatus status = event.getStatus();
        
        // 记录事务计数
        Metrics.counter("seata.transaction.count", 
                       Tags.of("name", transactionName, "status", status.name()))
               .increment();
        
        // 记录事务耗时
        if (event.getBeginTime() > 0 && event.getEndTime() > 0) {
            long duration = event.getEndTime() - event.getBeginTime();
            Metrics.timer("seata.transaction.duration", 
                         Tags.of("name", transactionName))
                   .record(duration, TimeUnit.MILLISECONDS);
        }
    }
    
    /**
     * 处理事务超时
     */
    private void handleTransactionTimeout(GlobalTransactionEvent event) {
        log.warn("分布式事务超时: xid={}, name={}", event.getId(), event.getName());
        
        // 发送告警
        TransactionTimeoutAlert alert = new TransactionTimeoutAlert();
        alert.setXid(event.getId());
        alert.setTransactionName(event.getName());
        alert.setStatus(event.getStatus().name());
        alert.setTimestamp(new Date());
        
        AsyncManager.me().execute(() -> {
            alertService.sendTransactionTimeoutAlert(alert);
        });
    }
}

总结

本文详细介绍了数据库连接与事务管理，包括：

1. HikariCP连接池优化：连接池配置、性能调优、监控告警
2. Spring事务管理：声明式事务、事务传播行为、编程式事务
3. 分布式事务处理：Seata AT/TCC/Saga模式、事务监控

这些技术为企业级应用提供了高性能的数据库连接管理和可靠的事务保证机制。

在下一篇文章中，我们将探讨数据库监控与性能调优。

参考资料

• HikariCP官方文档
• Spring事务管理文档
• Seata分布式事务文档