MySQL实现分布式锁：原理、步骤与代码示例

CREATE TABLE IF NOT EXISTS fud_distribute_lock (
    id BIGINT UNSIGNED PRIMARY KEY AUTO_INCREMENT,
    biz VARCHAR(50) COMMENT '业务Key',
    UNIQUE KEY uk_biz (biz)
) ENGINE=InnoDB;

// 省略了构造方法，需要传入DataSource和biz
private static final String SELECT_SQL = 
    "SELECT * FROM fud_distribute_lock WHERE `biz` = ? for update";
private static final String INSERT_SQL = 
    "INSERT INTO fud_distribute_lock(`biz`) values(?)";

// 从构造方法中传入
private final DataSource source;
private Connection connection;

public void lock() {
    PreparedStatement psmt = null;
    ResultSet rs = null;
    try {
        // 循环直到获取成功
        for (; ; ) {
            connection = this.source.getConnection();
            // 关闭自动提交事务，维持事务上下文以持有锁
            connection.setAutoCommit(false);
            
            psmt = connection.prepareStatement(SELECT_SQL);
            psmt.setString(1, biz);
            rs = psmt.executeQuery();
            
            // 如果查询到了数据，说明当前已有该业务的其他客户端持有锁，或者自己已获取
            if (rs.next()) {
                return; 
            }
            
            // 如果没有查到记录，需要先插入一行来“占据”这个位置
            // 因为for update只能锁定已经存在的行
            connection.commit(); // 先提交刚才的空事务
            close(connection, psmt, rs);
            
            Connection updConnection = this.source.getConnection();
            PreparedStatement insertStatement = null;
            try {
                insertStatement = updConnection.prepareStatement(INSERT_SQL);
                insertStatement.setString(1, biz);
                if (insertStatement.executeUpdate() == 1) {
                    LOGGER.info("创建锁记录成功");
                } else {
                    // 如果返回0，说明存在竞争，其他线程抢先插入了
                    continue;
                }
            } catch (Exception e) {
                // 如果发生死锁或其他异常导致插入失败，说明有竞争对手已持有该唯一键
                // 此时应当重试，重新进入循环去select...for update
                LOGGER.error("创建锁记录异常", e); 
                continue; 
            } finally {
                close(insertStatement, updConnection);
            }
            
            // 插入成功后，为了正式进入持有锁的状态，再次执行select for update
            psmt = connection.prepareStatement(SELECT_SQL);
            psmt.setString(1, biz);
            connection.setAutoCommit(false);
            rs = psmt.executeQuery();
            // 再次确认获取锁状态
            rs.next(); 
            return;
        }
    } catch (Exception e) {
        LOGGER.error("lock异常信息", e); 
        throw new RuntimeException(e);
    } finally {
        // 资源清理需注意保留当前连接connection供后续unlock使用
        if(psmt != null) { try { psmt.close(); } catch (Exception ignore) {} }
        if(rs != null) { try { rs.close(); } catch (Exception ignore) {} }
    }
}

public void unlock() {
    try {
        // 事务提交之后自动解锁（释放行上的X锁）
        connection.commit(); 
        close(connection);
    } catch (Exception e) {
        LOGGER.error("unlock异常信息", e); 
        throw new RuntimeException(e);
    }
}

public void close(AutoCloseable... closeables) {
    Arrays.stream(closeables).forEach(closeable -> {
        if (null != closeable) {
            try {
                closeable.close();
            } catch (Exception e) {
                LOGGER.error("close关闭异常", e);
            }
        }
    });
}