MYSQL隔离级别详解

MYSQL隔离级别详解

准备工作

创建一个测试数据库并创建一个简单的表。

mysql> create database mytest;
Query OK, 1 row affected (0.02 sec)

mysql> use mytest;
Database changed
mysql> show tables;
Empty set (0.00 sec)

mysql> create table test(
    -> id int auto_increment primary key,
    -> name varchar(50) not null,
    -> email varchar(50) not null unique
    -> );
Query OK, 0 rows affected (0.04 sec)

mysql> show tables;
+------------------+
| Tables_in_mytest |
+------------------+
| test             |
+------------------+
1 row in set (0.00 sec)

接下来我们使用python和sqlalchemy来演示不同的隔离级别的作用和影响。

SQLAlchemy 是一个用于 Python 的 SQL 工具包和对象关系映射（ORM）库。它提供了一种高效且灵活的方式来与数据库进行交互，支持多种数据库后端（如 SQLite、PostgreSQL、MySQL、Oracle 等）。

读未提交（Read Uncommitted）

• 最低的隔离级别。
• 事务可以读取其他事务尚未提交的数据（脏读）。
• 可能会出现脏读、不可重复读和幻读问题。

可以通过多线程的方式来看一下 read uncommitted的作用。

import time
import threading
from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, text
from  import sessionmaker

# 连接数据库
engine = create_engine('mysql+pymysql://user:passwd@localhost/mytest', isolation_level="READ UCOMMITTED")
metadata = MetaData()

# 确保表存在
test_table = Table('test', metadata,
                   Column('id', Integer, primary_key=True),
                   Column('name', String(50)),
                   Column('email', String(50)))

# 创建会话
Session = sessionmaker(bind=engine)

def insert_data():
    session = Session()
    try:
        session.begin()
        (test_table.insert().values(name="tom", email="xxx@qq"))
        sessionmit()
        print("Insert successful!")
    except Exception as e:
        session.rollback()
        print(f"Insert failed: {e}")
    finally:
        ()

def updata_data():
    session = Session()  # 每个线程创建自己的会话
    try:
        session.begin()
        (test_table.update().values(email="eee@qq").where(test_id == 1))
        time.sleep(5)  # 模拟长时间操作
        sessionmit()
        print("Update successful!")
    except Exception as e:
        session.rollback()
        print(f"Update failed: {e}")
    finally:
        ()

def read_data():
    session = Session()  # 每个线程创建自己的会话
    try:
        session.begin()
        result = (test_table.select()).fetchall()
        print("Read Uncommitted:", result)
    except Exception as e:
        session.rollback()
        print(f"Read failed: {e}")
    finally:
        ()

def main():
    # 插入一条数据
    insert_data()
    
    th1 = threading.Thread(target=updata_data)
    th1.start()
    time.sleep(1)  # 确保更新线程先启动
    th2 = threading.Thread(target=read_data)
    th2.start()

    th1.join()
    th2.join()


if __name__ == "__main__":
    main()

注意在创建数据库引擎的时候需要配置好隔离级别，这样在接下来创建的会话就都是同一隔离级别等级的了。

读数据的函数会在update完成commit之前就读取要更新的数据（id==1），因为我们设置了可以读取其他事务未提交的数据，所以这里应该可以看到读取的数据应该是eee@qq，随后update才完成。

(venv) yidaas@TOMMWWU-MC Scripts % python mytest.py
Insert successful!
Read Uncommitted: (1, 'tom', 'eee@qq')
Update successful!

读已提交（Read Committed）

• 事务只能读取其他事务已经提交的数据。
• 避免了脏读问题，但仍可能出现不可重复读和幻读问题。
• 每次读取数据时，都会看到其他事务已经提交的最新数据。

如果我将上面的代码修改一下隔离级别就可以看到我们不能再读取未commit的数据。

import time
import threading
from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, text
from  import sessionmaker

# 连接数据库
engine = create_engine('mysql+pymysql://user:passwd@localhost/mytest', isolation_level="READ COMMITTED")
metadata = MetaData()

# 确保表存在
test_table = Table('test', metadata,
                   Column('id', Integer, primary_key=True),
                   Column('name', String(50)),
                   Column('email', String(50)))

# 创建会话
Session = sessionmaker(bind=engine)

def updata_data():
    session = Session()  # 每个线程创建自己的会话
    try:
        session.begin()
        (test_table.update().values(email="ert@qq").where(test_id == 1))
        time.sleep(5)  # 模拟长时间操作
        sessionmit()
        print("Update successful!")
    except Exception as e:
        session.rollback()
        print(f"Update failed: {e}")
    finally:
        ()

def read_data():
    session = Session()  # 每个线程创建自己的会话
    try:
        session.begin()
        result = (test_table.select().where(test_id == 1)).first()
        print("Read Uncommitted:", result)
    except Exception as e:
        session.rollback()
        print(f"Read failed: {e}")
    finally:
        ()

def main():
    th1 = threading.Thread(target=updata_data)
    th1.start()
    time.sleep(1)  # 确保更新线程先启动
    th2 = threading.Thread(target=read_data)
    th2.start()

    th1.join()
    th2.join()


if __name__ == "__main__":
    main()

输出如下：可以看到我们读取的是update之前的数据，而不是要update的数据。

python mytest.py
Read Uncommitted: (1, 'tom', 'eee@qq')
Update successful!

可重复读（Repeatable Read）

• 事务在开始时，会创建一个数据快照，事务只能看到该快照中的数据。
• 避免了脏读和不可重复读问题，但可能会出现幻读问题。
• MySQL 的默认隔离级别。

在在可重复读事务开启的那一刻，该事务读取的都是那一刻的表的快照，也就是说，重复读读取的数据一直是一致的，它不像前两种，在事务（读未提交和读已提交）开启后读取的数据会受其他事务的影响。而可重复读。读取的是事务开启那一刻的数据快照（副本），所以在事务内无论何时读取到的都是一样的数据。

实验如下，我们在数据更新前，开启可重复读事务，并在事务内等待数据更新完成，观察数据是否变化，并在可重复事务结束后再次读取更新的数据，来验证结果的正确性。

import time
import threading
from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, text
from  import sessionmaker

# 连接数据库
engine = create_engine('mysql+pymysql://user:passwa@ip:06/mytest', isolation_level="Repeatable Read")
metadata = MetaData()

# 确保表存在
test_table = Table('test', metadata,
                   Column('id', Integer, primary_key=True),
                   Column('name', String(50)),
                   Column('email', String(50)))

# 创建会话
Session = sessionmaker(bind=engine)

def updata_data():
    session = Session()  # 每个线程创建自己的会话
    try:
        session.begin()
        (test_table.update().values(email="dsdfasf@qq").where(test_id == 1))
        sessionmit()
        print("Update successful!")
    except Exception as e:
        session.rollback()
        print(f"Update failed: {e}")
    finally:
        ()

def read_data():
    session = Session()  # 每个线程创建自己的会话
    try:
        session.begin()
        result = (test_table.select().where(test_id == 1)).first()
        print("Read 1:", result)
        time.sleep(6)
        result = (test_table.select().where(test_id == 1)).first()
        print("Read 2:", result)
    except Exception as e:
        session.rollback()
        print(f"Read failed: {e}")
    finally:
        ()

    # 可重复读事务结束后再次读取数据，看是否真的变化。
    result = (test_table.select().where(test_id == 1)).first()
    print("Read :", result)

def main():
    # 插入一条数据
    # insert_data()

    #先读取更新之前的数据，然后在事务内等待数据更新之后，再次读取数据，观察是否存在变化
    th1 = threading.Thread(target=read_data)
    th1.start()
    time.sleep(1) # 确保读线程先启动
    th2 = threading.Thread(target=updata_data)
    th2.start()

    th1.join()
    th2.join()


if __name__ == "__main__":
    main()

输出结果如下：

Read 1: (1, 'tom', 'qq@qq')
Update successful!
Read 2: (1, 'tom', 'qq@qq')
Read : (1, 'tom', 'dsdfasf@qq')

串行化（Serializable）

串行化的逻辑就很简单了，跟编程器中的串行化一样，保证事务都是按顺序执行，因此也就不可能存在脏读，幻读等数据问题了，不过会带来明显的性能下降，因为它会导致更多的锁定和等待。

• 避免脏读：一个事务不能读取另一个事务尚未提交的数据。
• 避免不可重复读：一个事务在读取某行数据后，其他事务不能修改该行数据，直到第一个事务完成。
• 避免幻读：一个事务在读取某个范围的数据后，其他事务不能在该范围内插入新数据，直到第一个事务完成。

总结

一般情况我们使用mysql默认的隔离级别就可以了，但是还是要根据自己的业务场景进行选择，下面给出一些不同隔离级别适用的可能场景：

• 低数据一致性要求：适用于对数据一致性要求不高的场景，例如某些日志记录或监控系统。
• 高性能需求：适用于需要极高并发性能的场景，因为这种隔离级别几乎没有锁定开销。

• 中等数据一致性要求：适用于大多数在线事务处理（OLTP）系统，例如银行系统、电子商务平台等。
• 避免脏读：适用于需要避免脏读但可以容忍不可重复读的场景。

• 高数据一致性要求：适用于需要高数据一致性但可以容忍一定性能开销的场景，例如金融交易系统。
• 避免脏读和不可重复读：适用于需要避免脏读和不可重复读的场景。

• 最高数据一致性要求：适用于需要最高数据一致性且可以容忍显著性能开销的场景，例如某些金融系统、库存管理系统等。
• 完全避免并发问题：适用于需要完全避免脏读、不可重复读和幻读的场景。