企业官网建设流程全解析

主键没查到，为什么INSERT还在等？把InnoDB锁画在10和20之间

前面那篇行锁实验里，A锁住已有的记录，B修改同一行就等待。这个好理解：这行正被另一个事务占着。

现在把条件改一下。表里只有主键10和20，A查15，返回空集。15这一行都不存在，B再插入15，总该能继续吧？

先不背“间隙锁”这个名字。我们把这条空查询留下，再开一个窗口，看看究竟发生了什么。

留着A的事务，去B插入一条记录

本次实测是MySQL8.0.46、InnoDB，A显式使用REPEATABLE READ（可重复读，后面简称RR）。id是单列主键，没有其他索引或外键。下面在自己的临时实例操作；数据库名已存在时换一个新名字。

任意会话先建表：

CREATE DATABASE gap_lab; USE gap_lab; CREATE TABLE account ( id INT PRIMARY KEY, balance INT NOT NULL ) ENGINE=InnoDB; INSERT INTO account VALUES (10,100), (20,200);

A执行到SELECT就停下，先别提交：

USE gap_lab; SELECT CONNECTION_ID(); SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ; START TRANSACTION; SELECT * FROM account WHERE id=15 FOR UPDATE;

屏幕上是空集。注意这里有FOR UPDATE，并且事务还没结束。

B也连同一个实例，执行：

USE gap_lab; SELECT CONNECTION_ID(); SET SESSION innodb_lock_wait_timeout=30; START TRANSACTION; INSERT INTO account VALUES (15,150);

这次INSERT没有马上返回。先别急着在B重复输入，去第三个会话C看一眼。C需要有读取performance_schema的权限：

SELECT t.PROCESSLIST_ID, l.INDEX_NAME, l.LOCK_MODE, l.LOCK_STATUS, l.LOCK_DATA FROM performance_schema.data_locks AS l JOIN performance_schema.threads AS t ON t.THREAD_ID=l.THREAD_ID WHERE l.OBJECT_SCHEMA='gap_lab' AND l.OBJECT_NAME='account' ORDER BY t.PROCESSLIST_ID, l.LOCK_DATA;

按刚才记下的连接ID区分A和B。本次关键的两条记录是下面这样，省略了表级意向锁：

A PRIMARY X,GAP GRANTED 20 B PRIMARY X,GAP,INSERT_INTENTION WAITING 20

两个LOCK_DATA都是20。明明查的是15，为什么这里却写20？

先把15要放的位置画出来

主键按顺序排列，现在只有两个位置放了记录：

已有记录 已有记录 10 20 |---------------| ↑ 15要插在这里

查15时找不到记录，但可以确定它应该在10与20之间。A持有的X,GAP描述的是这个间隙；在这次输出里，它附着在右端索引记录20上。

所以不能把LOCK_DATA=20直接读成“20这条业务记录被独占了”。还得一起看LOCK_MODE：这里有GAP。

A查询15：空集，事务未结束 10 [ 暂时不让新记录插进来 ] 20 ↑ B想把15放进这里 ↓ 等A结束

如果只挡住15这个值，那么插入12应该没问题。可以重新做一轮：先在A执行COMMIT，让B的INSERT返回；B执行ROLLBACK，撤销刚插入的15，使表回到只有10和20。

接着A重新开始同样的事务、同样查询15，B把插入值换成12。结果12也等待。19同样等待；换成25，在A还没提交时就能完成。每轮结束都在B回滚，以免上一轮改变下一轮的间隙。

现在范围就看出来了：挡住的是10与20之间的新插入，不只是查询条件里写出的那个15。25不落在这里，也不是“整张表都锁住了”。

为什么会顺带影响12？可以回到那张主键顺序图：当前索引里没有12或15的记录，二者要进入的是同一个相邻记录间隙。这里的锁以这个索引间隙为单位，不是在给业务条件中的每个整数单独建一把锁。

MySQL对间隙锁的说明把它用于阻止向相应索引间隙插入；间隙锁之间的兼容规则，也不能照搬记录共享锁、排他锁的直觉。

把查询改回10，会有什么区别？

还是每轮先把两边事务收好，恢复10和20。A查询的条件改成：

START TRANSACTION; SELECT * FROM account WHERE id=10 FOR UPDATE;

这次能拿到已有记录10,100。观察会话看到的关键记录是：

A PRIMARY X,REC_NOT_GAP GRANTED 10

此时B插入15可以完成；B更新10则会等待。

查询10，确实找到这一行： [10被锁住] --- 15可插入 --- 20 ↑ B更新10要等

“唯一等值查询只锁一条记录”这句话，不能省掉“确实定位到已有记录”等前提。这里用到的是完整单列主键；复合唯一索引只给部分列、二级非唯一索引、范围查询，都不能直接套这个图。

不加FOR UPDATE呢？换隔离级别呢？

A恢复REPEATABLE READ，开启事务，只运行普通SELECT：

START TRANSACTION; SELECT * FROM account WHERE id=15;

同样是空集，这次B插入15可以完成。本例普通SELECT是非锁定一致性读；不能看见一个SELECT就认为它把扫描范围锁住了。

再结束双方事务、恢复数据。A在新事务开始前改成READ COMMITTED（读已提交，简称RC）：

SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED; START TRANSACTION; SELECT * FROM account WHERE id=15 FOR UPDATE;

B插入15也能完成。注意修改隔离级别的那句放在START TRANSACTION之前，不是在原事务中途切换。

把这几轮放在一起读，才能知道是哪一个条件起了作用：

RC下针对搜索、扫描通常不采用同样的间隙锁机制，但外键检查、重复键检查还有例外。本文这张简单表的结果，不等于“RC绝不会出现间隙锁”。相应规则见MySQL隔离级别说明。

查询空集以后，要不要一直占着这个事务？

这个问题在“先查有没有，再决定插入”的代码里很实际。查询没有找到数据，不代表它没有影响别人；事务如果在这里等用户操作、调用远程接口，其他落在同一间隙的插入就可能一起等着。

反过来，间隙锁也不是“业务查重已经万无一失”的凭据。多个事务先查空、再同时插入，仍要考虑等待和死锁；最后防止重复数据的约束应由唯一索引等机制承担，并处理实际返回的错误。

排查一次INSERT卡住时，可以按这次的顺序走：先确认它真的在等锁，再找到持锁事务的SQL、隔离级别和未提交时间，最后结合索引顺序画出它要插的位置。先别因为SELECT返回空集，就把那个会话排除掉。

附：自动实验怎样避免“好像卡住了”的判断

人工实验适合看过程，自动程序则需要明确的先后顺序：A收到SELECT完成标记后才启动B；预期等待的分支必须在data_lock_waits中观察到B对A的等待关系，才让A提交。只用sleep等一秒，无法区分锁等待与机器慢。

预期不等待的分支则要求：A仍保持事务时，B已经收到SQL完成标记。每轮回滚B，确认表还是10,100与20,200，再开始下一轮。

下面是完整运行器。它新建临时数据目录，只使用Unix socket，不连接原有数据库。需要Linux、Python3和MySQL 8.0的二进制发行目录；通过BLOG_MYSQL_ROOT指定解包根目录，其中应包含usr/sbin/mysqld、usr/bin/mysql以及相应动态库。若你使用系统安装的MySQL，前面的三个会话SQL更适合直接复现。

保存为verify_gap.py，普通用户运行：

BLOG_MYSQL_ROOT=/path/to/mysql-package-root python3 verify_gap.py

结果写在程序旁的results.json；每个运行目录保留服务日志与客户端错误日志。临时实例关闭后保留实验目录，方便检查，没有自动删除已有目录。

"""Run only in an isolated MySQL instance; never connect to a user's database.""" import getpass import json import os from pathlib import Path import select import subprocess import tempfile import time base = Path(__file__).resolve().parent package = Path(os.environ.get('BLOG_MYSQL_ROOT', '/mnt/d/blog-ops/experiments/2026-10-03-candidates/mysql-packages/root')) env = dict(os.environ, LD_LIBRARY_PATH=str(package / 'usr/lib/x86_64-linux-gnu')) run = Path(tempfile.mkdtemp(prefix='gap-', dir=base)) (run / 'data').mkdir() sock = Path('/dev/shm') / (run.name + '.sock') args = [str(package / 'usr/sbin/mysqld'), '--no-defaults', '--basedir=' + str(package / 'usr'), '--datadir=' + str(run / 'data'), '--lc-messages-dir=' + str(package / 'usr/share/mysql'), '--user=' + getpass.getuser(), '--secure-file-priv=NULL', '--innodb-buffer-pool-size=67108864', '--innodb-redo-log-capacity=16777216', '--log-error=' + str(run / 'server.log')] subprocess.run(args + ['--initialize-insecure'], env=env, check=True, capture_output=True, timeout=90) client = [str(package / 'usr/bin/mysql'), '--no-defaults', '--protocol=SOCKET', '--socket=' + str(sock), '-u', 'root', '--batch', '--skip-column-names', '--unbuffered', '--connect-timeout=2'] server = subprocess.Popen(args + ['--skip-networking', '--mysqlx=0', '--socket=' + str(sock), '--pid-file=' + str(run / 'mysql.pid'), '--tmpdir=' + str(run)], env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) sessions = [] record = {'passed': False, 'run': str(run), 'cases': []} observer_sql = '''SELECT t.PROCESSLIST_ID, l.INDEX_NAME, l.LOCK_MODE, l.LOCK_STATUS, l.LOCK_DATA FROM performance_schema.data_locks AS l JOIN performance_schema.threads AS t ON t.THREAD_ID=l.THREAD_ID WHERE l.OBJECT_SCHEMA='gap_lab' AND l.OBJECT_NAME='account' ORDER BY t.PROCESSLIST_ID, l.LOCK_DATA;''' class Session: def __init__(self, name): self.err = open(run / (name + '.stderr'), 'wb') self.p = subprocess.Popen(client, env=env, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=self.err) self.buffer = b'' self.counter = 0 sessions.append(self) def send(self, text): self.counter += 1 self.marker = '__done_%d__' % self.counter self.p.stdin.write((text + "\nSELECT '%s';\n" % self.marker).encode()) self.p.stdin.flush() def receive(self): lines = [] deadline = time.monotonic() + 15 while time.monotonic() < deadline: while b'\n' in self.buffer: line, self.buffer = self.buffer.split(b'\n', 1) value = line.decode().rstrip('\r') if value == self.marker: return lines lines.append(value) if select.select([self.p.stdout], [], [], .1)[0]: data = os.read(self.p.stdout.fileno(), 65536) if not data: raise RuntimeError('MySQL client ended; see ' + self.err.name) self.buffer += data raise TimeoutError('SQL did not reach its completion marker') def sql(self, text): self.send(text) return self.receive() def close(self): if self.p.poll() is None: self.p.stdin.close() try: self.p.wait(timeout=5) except subprocess.TimeoutExpired: self.p.kill() self.p.wait() self.p.stdout.close() self.err.close() try: for _ in range(200): ping = subprocess.run(client, input='SELECT VERSION();', text=True, env=env, capture_output=True, timeout=3) if ping.returncode == 0: record['version'] = ping.stdout.strip() break if server.poll() is not None: raise RuntimeError((run / 'server.log').read_text()) time.sleep(.1) else: raise TimeoutError('server startup') a, b, c = (Session(n) for n in ['A', 'B', 'C']) c.sql('CREATE DATABASE gap_lab; USE gap_lab; ' 'CREATE TABLE account(id INT PRIMARY KEY,balance INT NOT NULL) ENGINE=InnoDB; ' 'INSERT INTO account VALUES(10,100),(20,200);') for s in (a, b): s.sql('USE gap_lab; SET SESSION innodb_lock_wait_timeout=30;') ids = [int(s.sql('SELECT CONNECTION_ID();')[0]) for s in (a, b, c)] record['connectionIds'] = ids # Same schema; change one condition at a time. cases = [ ('rr-missing-15', 'REPEATABLE READ', 15, True, 'INSERT INTO account VALUES(15,150);', True), ('rr-missing-12', 'REPEATABLE READ', 15, True, 'INSERT INTO account VALUES(12,120);', True), ('rr-missing-19', 'REPEATABLE READ', 15, True, 'INSERT INTO account VALUES(19,190);', True), ('rr-outside-25', 'REPEATABLE READ', 15, True, 'INSERT INTO account VALUES(25,250);', False), ('rc-missing-15', 'READ COMMITTED', 15, True, 'INSERT INTO account VALUES(15,150);', False), ('rr-existing-insert', 'REPEATABLE READ', 10, True, 'INSERT INTO account VALUES(15,150);', False), ('rr-existing-update', 'REPEATABLE READ', 10, True, 'UPDATE account SET balance=101 WHERE id=10;', True), ('rr-ordinary', 'REPEATABLE READ', 15, False, 'INSERT INTO account VALUES(15,150);', False), ] for name, isolation, key, locking, action, should_wait in cases: a.sql('SET SESSION TRANSACTION ISOLATION LEVEL ' + isolation + '; START TRANSACTION;') selection = f'SELECT * FROM account WHERE id={key}' + (' FOR UPDATE;' if locking else ';') selected = a.sql(selection) assert selected == (['10\t100'] if key == 10 else []), selected locks = c.sql("SELECT INDEX_NAME,LOCK_TYPE,LOCK_MODE,LOCK_STATUS,LOCK_DATA " "FROM performance_schema.data_locks WHERE OBJECT_SCHEMA='gap_lab' " "AND THREAD_ID=(SELECT THREAD_ID FROM performance_schema.threads " f"WHERE PROCESSLIST_ID={ids[0]}) ORDER BY INDEX_NAME,LOCK_DATA;") record_locks = [line for line in locks if line.startswith('PRIMARY\tRECORD\t')] if isolation == 'REPEATABLE READ' and locking: assert record_locks == (['PRIMARY\tRECORD\tX,GAP\tGRANTED\t20'] if key == 15 else ['PRIMARY\tRECORD\tX,REC_NOT_GAP\tGRANTED\t10']), locks else: assert not record_locks, locks b.sql('START TRANSACTION;') b.send(action) waits = [] if should_wait: deadline = time.monotonic() + 8 while time.monotonic() < deadline: waits = c.sql("SELECT r.LOCK_MODE,r.LOCK_STATUS,r.LOCK_DATA," "bl.LOCK_MODE,bl.LOCK_STATUS,bl.LOCK_DATA " "FROM performance_schema.data_lock_waits w " "JOIN performance_schema.data_locks r ON r.ENGINE_LOCK_ID=w.REQUESTING_ENGINE_LOCK_ID " "AND r.ENGINE=w.ENGINE " "JOIN performance_schema.data_locks bl ON bl.ENGINE_LOCK_ID=w.BLOCKING_ENGINE_LOCK_ID " "AND bl.ENGINE=w.ENGINE " "JOIN performance_schema.threads t ON t.THREAD_ID=w.REQUESTING_THREAD_ID " f"WHERE t.PROCESSLIST_ID={ids[1]};") if waits: break time.sleep(.05) assert waits, name + ' did not enter an observed lock wait' assert b.p.poll() is None observed = c.sql(observer_sql) assert any(line.startswith(str(ids[0])+'\tPRIMARY\t') for line in observed) assert any(line.startswith(str(ids[1])+'\tPRIMARY\t') and '\tWAITING\t' in line for line in observed), observed a.sql('COMMIT;') assert b.receive() == [] else: assert b.receive() == [] # A still holds its transaction here. observed = c.sql(observer_sql) a.sql('COMMIT;') b.sql('ROLLBACK;') assert c.sql('SELECT * FROM account ORDER BY id;') == ['10\t100', '20\t200'] record['cases'].append({'id': name, 'isolation': isolation, 'select': selection, 'selected': selected, 'action': action, 'waitObserved': bool(waits), 'locks': locks, 'waits': waits, 'observerRows': observed, 'completedAfterACommit': should_wait}) record['passed'] = True finally: for s in sessions: s.close() server.terminate() try: server.wait(timeout=30) except subprocess.TimeoutExpired: server.kill() server.wait() record['serverStopped'] = server.poll() is not None record['clientsStopped'] = all(s.p.poll() is not None for s in sessions) record['socketRemoved'] = not sock.exists() (base / 'results.json').write_text(json.dumps(record, ensure_ascii=False, indent=2)) print('RESULTS=' + str(base / 'results.json'))