## 模块 7：使用集群分支加速创新

![demo_203-tidb-x-branching.jpg 85%](https://lab-static.pingcap.com/images/2025/11/27/c5caa8e851e261578fb8ca4576db01916c173685.jpg)

> [!NOTE]
> 本模块介绍的功能当前处于预览阶段。

本模块演示了分支功能如何显著提升开发的效率。

分支是一个独立的集群实例，包含从原始集群分出的自己的数据副本。这个隔离的环境可以让你安全地尝试和测试变更，不会影响主集群。在本模块中，你将从当前数据库创建三个独立的分支，并在每个分支上并行测试三种不同的库存解决方案。

开发者希望测试三种库存扣减方案：

- 方案 1：在单个事务内，先检查剩余库存，使用 “… FOR UPDATE” 锁定库存行，然后创建订单并扣减库存。
- 方案 2：将操作分为三个独立事务分别执行：先检查库存，再创建订单，最后扣减库存。
- 方案 3：在单个事务内检查剩余库存、创建订单并扣减库存（不加行级锁）。

为了评估每种方案的正确性和性能，为每种方案分别分配一个独立分支，并行运行测试。目标是确认哪种方案能防止“超卖”（即订单数量超过可用库存，说明逻辑有缺陷）。

### 步骤 1. 创建分支

- 在你的 TiDB Cloud 控制台，进入 [集群页面](https://tidbcloud.com/project/clusters)，点击你的实验集群名称。如果你通过 `创建新集群` 开始实验，集群名为 `${LAB:SERVERLESS_CLUSTER_NAME}`。

![Demo-RU-001.png 85%](https://lab-static.pingcap.com/images/2025/11/24/9c016ddc77e89cb9d9ae0669890d36cb316f6d20.png)

- 点击集群名称 `${LAB:SERVERLESS_CLUSTER_NAME}`，接着在左上角选择 `Overview`（概览）。

![Demo-RU-002.png 85%](https://lab-static.pingcap.com/images/2025/11/24/8d064cdda27759df1640171cbf92f7d340e4364d.png)

- 点击左侧的 `Branches`（分支）标签以打开分支面板。

![demo_branch_UI_01.png 85%](https://lab-static.pingcap.com/images/2025/11/27/c094e9effcea9e16423f52268a3f03be1d9f27fe.png)

- 在分支面板右上方点击 `Create Branch`（创建分支）按钮，新建一个分支。

![demo_branch_UI_02.png 85%](https://lab-static.pingcap.com/images/2025/11/27/544986b0dc326f95d55e8674dd3b403630801592.png)

- 在弹出的窗口中，Name 字段填 `test-branch-1`，Parent branch 选择 `main`，Include data up to 字段选择 `Current point in time`（当前时点），即以当前数据库的快照创建名为 `test-branch-1` 的分支。然后点击创建按钮。

![demo_branch_UI_03.png 50%](https://lab-static.pingcap.com/images/2025/11/27/1a617fd5c838b7b4559866a7196edf6fb4b71aaf.png)

- 以同样方式新建名为 `test-branch-2` 的分支，快照点同样为当前时点。

![demo_branch_UI_04.png 50%](https://lab-static.pingcap.com/images/2025/11/27/f14658cf6af60b5bebde97b0c9baeb72196855a0.png)

- 以同样方式新建名为 `test-branch-3` 的分支，快照点同样为当前时点。

![demo_branch_UI_05.png 50%](https://lab-static.pingcap.com/images/2025/11/27/067cc091c2c27bb173c4b1d47ecaefee42b79e5a.png)

- 返回分支面板，查看这三个新创建的分支，并关注 Status 列的状态。

![demo_branch_UI_06.png 85%](https://lab-static.pingcap.com/images/2025/11/27/5719c270a2536174780784e9ec2c970d0baa51ef.png)

- 等待这三个分支在 Status 列显示为 `Active`（活跃）后，说明所有分支已准备就绪。

![demo_branch_UI_07.png 85%](https://lab-static.pingcap.com/images/2025/11/27/7316875a385eb9453c8d5985677bdb5739f1c1e0.png)

### 步骤 2. 连接到所有分支

- 在分支面板中，选择 `test-branch-1`，点击操作栏 `...`，然后选择 `Connect`（连接）。

![demo_branch_UI_08.png 85%](https://lab-static.pingcap.com/images/2025/11/27/fcc16837cc8a9500b11dfce73d167fa16664ba49.png)

- 在弹窗中点击 `Generate Password`（生成密码）。

![demo_branch_UI_09.png 85%](https://lab-static.pingcap.com/images/2025/11/27/acda36b4b8b64f0e63910a71ee3a51bca663a969.png)

- 生成密码后，选择 `Connect with .env` 并点击 `copy`（复制）图标。

![env-file-branch-1.jpg 50%](https://lab-static.pingcap.com/images/2025/12/2/d16d8e8bca04b77d597b2df188502031c49dd771.jpg)

- 在 [终端 1](ssh://monitor1-ssh#1)，粘贴复制的 `test-branch-1` 连接信息以设置环境变量。结果类似如下：

  ```output
  ubuntu@ip-10-90-1-65:~$ DB_HOST=gateway01.us-west-2.prod.aws.tidbcloud.com
  DB_PORT=4000
  DB_USERNAME='3NViMyHWFBkJvQq.root'
  DB_PASSWORD='1Pbcc4CFXTtnZA4C'
  DB_DATABASE='lab_demo_203_6930011'
  ```

- 保持在 [终端 1](ssh://monitor1-ssh#1)，执行下面命令，将 `test-branch-1` 的连接信息保存到 `.env-1` 文件中。

  ```bash
  cat > .env-1 <<EOF
  BRANCH_HOST=$DB_HOST
  BRANCH_PORT=$DB_PORT
  BRANCH_USERNAME=$DB_USERNAME
  BRANCH_PASSWORD=$DB_PASSWORD
  BRANCH_DATABASE=${LAB:SERVERLESS_CLUSTER_DATABASE_NAME}
  EOF
  ```

- 按照上面的方法为 `test-branch-2` 创建连接信息，先生成密码，再选择 `Connect with .env` 并点击复制图标。

![env-file-branch-2.jpg 50%](https://lab-static.pingcap.com/images/2025/12/2/3eac0b2ee150121143fe91b4b514bbecc617e5fc.jpg)

- 在 [终端 2](ssh://monitor1-ssh#2)，粘贴复制的 `test-branch-2` 连接信息以设置环境变量。示例结果如下：

  ```output
  ubuntu@ip-10-90-1-65:~$ DB_HOST=gateway01.us-west-2.prod.aws.tidbcloud.com
  DB_PORT=4000
  DB_USERNAME='zxBRD32X3wj8KRY.root'
  DB_PASSWORD='PvNnqFGh63rAX4RK'
  DB_DATABASE='lab_demo_203_6930011'
  ```

- 保持在 [终端 2](ssh://monitor1-ssh#2)，执行下面命令，将 `test-branch-2` 的环境变量保存到 `.env-2` 文件中。

  ```bash
  cat > .env-2 <<EOF
  BRANCH_HOST=$DB_HOST
  BRANCH_PORT=$DB_PORT
  BRANCH_USERNAME=$DB_USERNAME
  BRANCH_PASSWORD=$DB_PASSWORD
  BRANCH_DATABASE=${LAB:SERVERLESS_CLUSTER_DATABASE_NAME}
  EOF
  ```

- 按照相同步骤为 `test-branch-3` 创建连接信息，先生成密码，再选择 `Connect with .env` 并点击复制图标。

![env-file-branch-3.jpg 50%](https://lab-static.pingcap.com/images/2025/12/2/2c0584b8494b0cfaf347cc86203f0fb991e1d032.jpg)

- 在 [终端 3](ssh://monitor1-ssh#3)，粘贴复制的 `test-branch-3` 连接信息进行环境变量赋值，示例结果如下：

  ```output
  ubuntu@ip-10-90-1-65:~$ DB_HOST=gateway01.us-west-2.prod.aws.tidbcloud.com
  DB_PORT=4000
  DB_USERNAME='2SE7X5kxT7oxJU5.root'
  DB_PASSWORD='AlrsLuPTxy76zFdI'
  DB_DATABASE='lab_demo_203_6930011'
  ```

- 保持在 [终端 3](ssh://monitor1-ssh#3)，执行下面命令，将 `test-branch-3` 连接信息保存到 `.env-3` 文件。

  ```bash
  cat > .env-3 <<EOF
  BRANCH_HOST=$DB_HOST
  BRANCH_PORT=$DB_PORT
  BRANCH_USERNAME=$DB_USERNAME
  BRANCH_PASSWORD=$DB_PASSWORD
  BRANCH_DATABASE=${LAB:SERVERLESS_CLUSTER_DATABASE_NAME}
  EOF
  ```

- 在任意终端，确认三个环境变量文件均已正确包含各分支的连接信息，你应该能看到三个不同的用户名。

  ```bash
  cat .env-1 .env-2 .env-3 | grep BRANCH_USERNAME
  ```

  ```output
  BRANCH_USERNAME=XXXXXXXXXXXXXXX.root
  BRANCH_USERNAME=YYYYYYYYYYYYYYY.root
  BRANCH_USERNAME=ZZZZZZZZZZZZZZZ.root
  ```

- 在 [终端 1](ssh://monitor1-ssh#1)，连接到 `test-branch-1`。

  ```bash
  source .env-1
  mysql \
  --comments -u $BRANCH_USERNAME -p$BRANCH_PASSWORD \
   -h $BRANCH_HOST -P $BRANCH_PORT \
   --ssl-mode=VERIFY_IDENTITY --ssl-ca=/etc/ssl/certs/ca-certificates.crt
  ```

  ```output
  mysql>
  ```

- 在 `test-branch-1` 中，将事务隔离级别设置为已提交读（READ-COMMITTED）。

  ```sql
  SET GLOBAL transaction_isolation = 'READ-COMMITTED';
  SHOW GLOBAL VARIABLES LIKE 'transaction_isolation';
  EXIT
  ```

  ```output
  Query OK, 0 rows affected (0.18 sec)

  +-----------------------+----------------+
  | Variable_name         | Value          |
  +-----------------------+----------------+
  | transaction_isolation | READ-COMMITTED |
  +-----------------------+----------------+
  1 row in set (0.16 sec)

  Bye
  ```

- 在 [终端 2](ssh://monitor1-ssh#2)，连接到 `test-branch-2`。

  ```bash
  source .env-2
  mysql \
  --comments -u $BRANCH_USERNAME -p$BRANCH_PASSWORD \
   -h $BRANCH_HOST -P $BRANCH_PORT \
   --ssl-mode=VERIFY_IDENTITY --ssl-ca=/etc/ssl/certs/ca-certificates.crt
  ```

  ```output
  mysql>
  ```

- 在 `test-branch-2` 中，同样设置事务隔离级别为已提交读。

  ```sql
  SET GLOBAL transaction_isolation = 'READ-COMMITTED';
  SHOW GLOBAL VARIABLES LIKE 'transaction_isolation';
  EXIT
  ```

  ```output
  Query OK, 0 rows affected (0.18 sec)

  +-----------------------+----------------+
  | Variable_name         | Value          |
  +-----------------------+----------------+
  | transaction_isolation | READ-COMMITTED |
  +-----------------------+----------------+
  1 row in set (0.16 sec)

  Bye
  ```

- 在 [终端 3](ssh://monitor1-ssh#3)，连接到 `test-branch-3`。

  ```bash
  source .env-3
  mysql \
  --comments -u $BRANCH_USERNAME -p$BRANCH_PASSWORD \
   -h $BRANCH_HOST -P $BRANCH_PORT \
   --ssl-mode=VERIFY_IDENTITY --ssl-ca=/etc/ssl/certs/ca-certificates.crt
  ```

  ```output
  mysql>
  ```

- 在 `test-branch-3` 中，同样设置事务隔离级别为已提交读。

  ```sql
  SET GLOBAL transaction_isolation = 'READ-COMMITTED';
  SHOW GLOBAL VARIABLES LIKE 'transaction_isolation';
  EXIT
  ```

  ```output
  Query OK, 0 rows affected (0.18 sec)

  +-----------------------+----------------+
  | Variable_name         | Value          |
  +-----------------------+----------------+
  | transaction_isolation | READ-COMMITTED |
  +-----------------------+----------------+
  1 row in set (0.16 sec)

  Bye
  ```

### 步骤 3. 在三个分支上并发部署测试

- 现在有三个可用分支，你可以同时并发测试三种不同的库存扣减方案。

- 请确保已打开三个终端，[终端 1](ssh://monitor1-ssh#1)、[终端 2](ssh://monitor1-ssh#2)、[终端 3](ssh://monitor1-ssh#3)。

- 请确保已为各分支准备好环境变量文件。你已在步骤 2 中完成此任务。可以在任意终端运行如下命令检查文件：

  ```bash
  file .env-1 .env-2 .env-3
  ```

  ```output
  .env-1: ASCII text
  .env-2: ASCII text
  .env-3: ASCII text
  ```

- 部署第一份 Python 代码。本测试模拟 8 个并发会话购买同一商品：
  - 使用 `SELECT ... FOR UPDATE` 在 `book` 表中检查库存并锁定记录。
  - 向 `order` 表插入订单。
  - 从 `book` 表中扣减库存。
  - 所有操作在一个事务中完成。

![demo_branch_pic_02.png 50%](https://lab-static.pingcap.com/images/2025/11/27/5dacb92501bf6389d1e7b7c7a2f857801cd2b999.png)

- 在 [终端 1](ssh://monitor1-ssh#1)，生成 Python 代码文件 `demo_branch_1_test.py`。

  ```bash
  cat > demo_branch_1_test.py <<'EOF'
  import mysql.connector
  from prettytable import PrettyTable
  import datetime
  import time
  import random
  import threading
  import sys,os
  from dotenv import load_dotenv

  load_dotenv(sys.argv[1])
  branch_host = os.environ.get("BRANCH_HOST")
  branch_port = os.environ.get("BRANCH_PORT")
  branch_username = os.environ.get("BRANCH_USERNAME")
  branch_password = os.environ.get("BRANCH_PASSWORD")
  branch_database = '${LAB:SERVERLESS_CLUSTER_DATABASE_NAME}'

  config = {
    'user': branch_username,
    'password': branch_password,
    'host': branch_host,
    'database': branch_database,
    'raise_on_warnings': False,
    'port': branch_port,
  }

  # Simulate order and inventory processing for a connection, which continuously runs purchase simulations until inventory is depleted.
  def process_order_worker(user_id):
        # Establishing the connection
        conn = mysql.connector.connect(**config, buffered=True)

        try:
            while True:
                cursor = conn.cursor()

                # Enable transaction handling where each purchase activity is completed within a single transaction.
                cursor.execute("BEGIN")

                # Query the inventory of the product to be purchased and apply a lock on the product data.
                cursor.execute("""
                     SELECT stock FROM books
                     WHERE id = 6 FOR UPDATE
                 """)

                result = cursor.fetchone()

                # Check if the product is sold out.
                if not result or result[0] <= 0:
                    print(f"Stock is not enough, Stop!")
                    break

                current_stock = result[0]

                # Generate a random purchase quantity.
                purchase_quantity = random.randint(1, 10)

                if purchase_quantity > current_stock:
                    purchase_quantity = current_stock

                # Create an order record.
                cursor.execute("""
                     INSERT INTO orders(book_id, user_id, quality, ordered_at)
                     VALUES(6, %s, %s, CURRENT_TIMESTAMP)
                               """, (user_id, purchase_quantity,))

                # Reduce inventory.
                cursor.execute("""
                     UPDATE books
                     SET stock = stock - %s
                     WHERE id = 6
                           """,(purchase_quantity,))

                # If the purchase is successful, commit the transaction.
                conn.commit()

                print(f"User {user_id} successfully purchased {purchase_quantity} books, and the current remaining stock is {current_stock - purchase_quantity}")

                time.sleep(random.uniform(0.1, 0.5))

        except Exception as e:
            print(f"User {user_id} got error when process the order.")
            conn.rollback()
        finally:
            conn.close

  def main():

    # Simulate 8 sessions performing purchases concurrently.
    num_threads = 8

    start_time = 0
    end_time = 0

    print(f"Starting {num_threads} database worker threads...")

    # Create and start threads
    threads = []
    for i in range(num_threads):
        thread = threading.Thread(
            target=process_order_worker,
            args=(i + 1,),
            name=f"User-{i+1}",
            daemon=True
        )
        threads.append(thread)
        thread.start()
        print(f"{thread.name} thread started successfully!")

    print(f"All {num_threads} user threads started successfully!")

    start_time = time.time()

    try:
        # Keep the main thread alive - wait for threads to complete (they run indefinitely)
        for thread in threads:
            thread.join()

        end_time = time.time()

        conn = mysql.connector.connect(**config, buffered=True)

        # Print test results and test duration.
        try:
             cursor = conn.cursor()

             cursor.execute("""
                 SELECT stock FROM books
                 WHERE id = 6
             """)
             final_stock = cursor.fetchone()[0]

             cursor.execute("""
                 SELECT COUNT(*) as total_orders FROM orders WHERE book_id = 6
             """)
             total_orders = cursor.fetchone()[0]

             cursor.execute("""
                 SELECT SUM(quality) as total_quantity FROM orders WHERE book_id = 6
             """)
             total_quantity = cursor.fetchone()[0]

             duration_time = end_time - start_time

             print("=======================================")
             print(f"Final stock: {final_stock}")
             print(f"Total orders: {total_orders}")
             print(f"Total purchased books: {total_quantity}")
             print(f"Process duration: {duration_time:.2f} seconds.")
             print("Test finished！")
             print("=======================================")

        except Exception as e:
            print(f"Got error when get summary!")
            conn.rollback()
        finally:
            conn.close

    except KeyboardInterrupt:
        print("\nReceived interrupt signal. Stopping all threads...")
        sys.exit(0)

  if __name__ == "__main__":
    main()
  EOF
  ls -l demo_branch_1_test.py
  ```

  ```output
  -rw-rw-r-- 1 ubuntu ubuntu 4770 Nov 26 10:37 demo_branch_1_test.py
  ```

- 部署第二份用于测试的 Python 代码。本测试模拟 8 个并发会话以不同的事务的方式购买同一商品：
  - 检查库存（事务1）
  - 插入订单记录（事务2）
  - 扣减库存（事务3）

![demo_branch_pic_03.png 50%](https://lab-static.pingcap.com/images/2025/11/27/5c23d6698ceb6107676abbec140d556fa588e914.png)

- 在 [终端 2](ssh://monitor1-ssh#2)，生成 Python 代码 `demo_branch_2_test.py`。

  ```bash
  cat > demo_branch_2_test.py <<'EOF'
  import mysql.connector
  from prettytable import PrettyTable
  import datetime
  import time
  import random
  import threading
  import sys,os
  from dotenv import load_dotenv

  load_dotenv(sys.argv[1])
  branch_host = os.environ.get("BRANCH_HOST")
  branch_port = os.environ.get("BRANCH_PORT")
  branch_username = os.environ.get("BRANCH_USERNAME")
  branch_password = os.environ.get("BRANCH_PASSWORD")
  branch_database = '${LAB:SERVERLESS_CLUSTER_DATABASE_NAME}'

  config = {
    'user': branch_username,
    'password': branch_password,
    'host': branch_host,
    'database': branch_database,
    'raise_on_warnings': False,
    'port': branch_port,
  }

  # Simulate order and inventory processing for a connection, which continuously runs purchase simulations until inventory is depleted.
  def process_order_worker(user_id):

        # Establishing the connection
        conn = mysql.connector.connect(**config, buffered=True)

        try:
            while True:
                cursor = conn.cursor()

                # transaction handling where each purchase activity is completed across multiple transactions.
                cursor.execute("BEGIN")

                # Transaction 1: Query the inventory of the product to be purchased.
                cursor.execute("""
                     SELECT stock FROM books
                     WHERE id = 6
                 """)

                result = cursor.fetchone()

                # Check if the product is sold out.
                if not result or result[0] <= 0:
                    print(f"Stock is not enough, Stop!")
                    break

                current_stock = result[0]

                # Generate a random purchase quantity.
                purchase_quantity = random.randint(1, 10)

                if purchase_quantity > current_stock:
                    purchase_quantity = current_stock

                conn.commit()

                # Transaction 2: Create an order record.
                cursor.execute("BEGIN")

                # Reduce inventory.
                cursor.execute("""
                     INSERT INTO orders(book_id, user_id, quality, ordered_at)
                     VALUES(6, %s, %s, CURRENT_TIMESTAMP)
                               """, (user_id, purchase_quantity,))
                conn.commit()

                # Transaction 3: Reduce inventory.
                cursor.execute("BEGIN")

                cursor.execute("""
                     UPDATE books
                     SET stock = stock - %s
                     WHERE id = 6
                           """,(purchase_quantity,))

                # If the purchase is successful, commit the transaction.
                conn.commit()

                print(f"User {user_id} successfully purchased {purchase_quantity} books, and the current remaining stock is {current_stock - purchase_quantity}")

                time.sleep(random.uniform(0.1, 0.5))

        except Exception as e:
            print(f"User {user_id} got error when process the order.")
            conn.rollback()
        finally:
            conn.close

  def main():

    # Simulate 8 sessions performing purchases concurrently.
    num_threads = 8

    start_time = 0
    end_time = 0

    print(f"Starting {num_threads} database worker threads...")

    # Create and start threads
    threads = []
    for i in range(num_threads):
        thread = threading.Thread(
            target=process_order_worker,
            args=(i + 1,),
            name=f"User-{i+1}",
            daemon=True
        )
        threads.append(thread)
        thread.start()
        print(f"{thread.name} thread started successfully!")

    print(f"All {num_threads} user threads started successfully!")

    start_time = time.time()

    # Print test results and test duration.
    try:
        # Keep the main thread alive - wait for threads to complete (they run indefinitely)
        for thread in threads:
            thread.join()

        end_time = time.time()

        conn = mysql.connector.connect(**config, buffered=True)

        try:
             cursor = conn.cursor()

             cursor.execute("""
                 SELECT stock FROM books
                 WHERE id = 6
             """)
             final_stock = cursor.fetchone()[0]

             cursor.execute("""
                 SELECT COUNT(*) as total_orders FROM orders WHERE book_id = 6
             """)
             total_orders = cursor.fetchone()[0]

             cursor.execute("""
                 SELECT SUM(quality) as total_quantity FROM orders WHERE book_id = 6
             """)
             total_quantity = cursor.fetchone()[0]

             duration_time = end_time - start_time

             print("=======================================")
             print(f"Final stock: {final_stock}")
             print(f"Total orders: {total_orders}")
             print(f"Total purchased books: {total_quantity}")
             print(f"Process duration: {duration_time:.2f} seconds.")
             print("Test finished！")
             print("=======================================")

        except Exception as e:
            print(f"Got error when get summary!")
            conn.rollback()
        finally:
            conn.close

    except KeyboardInterrupt:
        print("\nReceived interrupt signal. Stopping all threads...")
        sys.exit(0)

  if __name__ == "__main__":
    main()
  EOF
  ls -l demo_branch_2_test.py
  ```

  ```output
  -rw-rw-r-- 1 ubuntu ubuntu 4922 Nov 26 10:54 demo_branch_2_test.py
  ```

- 部署第三份用于测试的 Python 代码。本测试模拟 8 个并发会话购买同一商品：
  - 不加锁检查库存。
  - 向订单表插入订单记录。
  - 从 `book` 表中扣减库存。
  - 所有操作在一个事务内完成。

![demo_branch_pic_04.png 50%](https://lab-static.pingcap.com/images/2025/11/27/f5b72da437ba5da5fbbc597a5e1fe8cba271e8f9.png)

- 在 [终端 3](ssh://monitor1-ssh#3)，生成 Python 代码 `demo_branch_3_test.py`。

  ```bash
  cat > demo_branch_3_test.py <<'EOF'
  import mysql.connector
  from prettytable import PrettyTable
  import datetime
  import time
  import random
  import threading
  import sys,os
  from dotenv import load_dotenv

  load_dotenv(sys.argv[1])
  branch_host = os.environ.get("BRANCH_HOST")
  branch_port = os.environ.get("BRANCH_PORT")
  branch_username = os.environ.get("BRANCH_USERNAME")
  branch_password = os.environ.get("BRANCH_PASSWORD")
  branch_database = '${LAB:SERVERLESS_CLUSTER_DATABASE_NAME}'

  config = {
    'user': branch_username,
    'password': branch_password,
    'host': branch_host,
    'database': branch_database,
    'raise_on_warnings': False,
    'port': branch_port,
  }

  # Simulate order and inventory processing for a connection, which continuously runs purchase simulations until inventory is depleted.
  def process_order_worker(user_id):

        # Establishing the connection
        conn = mysql.connector.connect(**config, buffered=True)

        try:
            while True:
                cursor = conn.cursor()

                # Enable transaction handling where each purchase activity is completed within a single transaction.
                cursor.execute("BEGIN")

                # Query the inventory of the product to be purchased without applying a lock on the product data.
                cursor.execute("""
                     SELECT stock FROM books
                     WHERE id = 6
                 """)

                result = cursor.fetchone()

                # Check if the product is sold out.
                if not result or result[0] <= 0:
                    print(f"Stock is not enough, Stop!")
                    break

                current_stock = result[0]

                # Generate a random purchase quantity.
                purchase_quantity = random.randint(1, 10)

                if purchase_quantity > current_stock:
                    purchase_quantity = current_stock

                # Create an order record.
                cursor.execute("""
                     INSERT INTO orders(book_id, user_id, quality, ordered_at)
                     VALUES(6, %s, %s, CURRENT_TIMESTAMP)
                               """, (user_id, purchase_quantity,))

                # Reduce inventory.
                cursor.execute("""
                     UPDATE books
                     SET stock = stock - %s
                     WHERE id = 6
                           """,(purchase_quantity,))

                conn.commit()

                print(f"User {user_id} successfully purchased {purchase_quantity} books, and the current remaining stock is {current_stock - purchase_quantity}")

                time.sleep(random.uniform(0.1, 0.5))

        except Exception as e:
            print(f"User {user_id} got error when process the order.")
            conn.rollback()
        finally:
            conn.close

  def main():

    # Simulate 8 sessions performing purchases concurrently.
    num_threads = 8

    start_time = 0
    end_time = 0

    print(f"Starting {num_threads} database worker threads...")

    # Create and start threads
    threads = []
    for i in range(num_threads):
        thread = threading.Thread(
            target=process_order_worker,
            args=(i + 1,),
            name=f"User-{i+1}",
            daemon=True
        )
        threads.append(thread)
        thread.start()
        print(f"{thread.name} thread started successfully!")

    print(f"All {num_threads} user threads started successfully!")

    start_time = time.time()

    try:
        # Keep the main thread alive - wait for threads to complete (they run indefinitely)
        for thread in threads:
            thread.join()

        end_time = time.time()

        conn = mysql.connector.connect(**config, buffered=True)

        # Print test results and test duration.
        try:
             cursor = conn.cursor()

             cursor.execute("""
                 SELECT stock FROM books
                 WHERE id = 6
             """)
             final_stock = cursor.fetchone()[0]

             cursor.execute("""
                 SELECT COUNT(*) as total_orders FROM orders WHERE book_id = 6
             """)
             total_orders = cursor.fetchone()[0]

             cursor.execute("""
                 SELECT SUM(quality) as total_quantity FROM orders WHERE book_id = 6
             """)
             total_quantity = cursor.fetchone()[0]

             duration_time = end_time - start_time

             print("=======================================")
             print(f"Final stock: {final_stock}")
             print(f"Total orders: {total_orders}")
             print(f"Total purchased books: {total_quantity}")
             print(f"Process duration: {duration_time:.2f} seconds.")
             print("Test finished！")
             print("=======================================")

        except Exception as e:
            print(f"Got error when get summary!")
            conn.rollback()
        finally:
            conn.close

    except KeyboardInterrupt:
        print("\nReceived interrupt signal. Stopping all threads...")
        sys.exit(0)

  if __name__ == "__main__":
    main()
  EOF
  ls -l demo_branch_3_test.py
  ```

  ```output
  -rw-rw-r-- 1 ubuntu ubuntu 4682 Nov 26 11:05 demo_branch_3_test.py
  ```

### 步骤 4. 运行并发测试

- 在 [终端 1](ssh://monitor1-ssh#1) 上运行 `demo_branch_1_test.py`（无需等待，直接继续）。

  ```bash
  ~/.venv/bin/python demo_branch_1_test.py .env-1
  ```

  ```output
  Starting 8 database worker threads...
  User-1 thread started successfully!
  User-2 thread started successfully!
  User-3 thread started successfully!
  User-4 thread started successfully!
  User-5 thread started successfully!
  User-6 thread started successfully!
  User-7 thread started successfully!
  User-8 thread started successfully!
  All 8 user threads started successfully!
  User 1 successfully purchased 3 books, and
  ...
  ```

- 在 [终端 2](ssh://monitor1-ssh#2) 上运行 `demo_branch_2_test.py`（无需等待，直接继续）。

  ```bash
  ~/.venv/bin/python demo_branch_2_test.py .env-2
  ```

  ```output
  Starting 8 database worker threads...
  User-1 thread started successfully!
  User-2 thread started successfully!
  User-3 thread started successfully!
  User-4 thread started successfully!
  User-5 thread started successfully!
  User-6 thread started successfully!
  User-7 thread started successfully!
  User-8 thread started successfully!
  All 8 user threads started successfully!
  User 7 successfully purchased 1 books, and
  ...
  ```

- 在 [终端 3](ssh://monitor1-ssh#3) 上运行 `demo_branch_3_test.py`（无需等待，直接继续）。

  ```bash
  ~/.venv/bin/python demo_branch_3_test.py .env-3
  ```

  ```output
  Starting 8 database worker threads...
  User-1 thread started successfully!
  User-2 thread started successfully!
  User-3 thread started successfully!
  User-4 thread started successfully!
  User-5 thread started successfully!
  User-6 thread started successfully!
  User-7 thread started successfully!
  User-8 thread started successfully!
  All 8 user threads started successfully!
  ...
  ```

- 监控所有三个测试直到结束。切换回 [终端 3](ssh://monitor1-ssh#3)（`test-branch-3`），该方案执行时间最短，但库存出现小于零的情况（超卖）——说明方案存在错误。注意：你的实验结果可能和本手册不同。如果初次没有出现超卖情况，请多运行几次测试。

  ```output
  ...
  User 4 successfully purchased 4 books, and the current remaining stock is 23
  User 7 successfully purchased 7 books, and the current remaining stock is 15
  User 6 successfully purchased 8 books, and the current remaining stock is 5
  Stock is not enough, Stop!
  Stock is not enough, Stop!
  User 2 successfully purchased 8 books, and the current remaining stock is 5
  User 3 successfully purchased 6 books, and the current remaining stock is 7
  Stock is not enough, Stop!
  User 8 successfully purchased 2 books, and the current remaining stock is 0
  Stock is not enough, Stop!
  Stock is not enough, Stop!
  Stock is not enough, Stop!
  Stock is not enough, Stop!
  Stock is not enough, Stop!
  =======================================
  Final stock: -22
  Total orders: 370
  Total purchased books: 2022
  Process duration: 68.15 seconds.
  Test finished！
  =======================================
  ```

- 回到 [终端 2](ssh://monitor1-ssh#2)（`test-branch-2`），该方案执行时间适中，但可能会出现库存超卖的情况，说明方案存在缺陷。你的实验结果可能与本手册不同。如果初次运行时未出现超卖，请多运行几次测试。

  ```output
  ...
  User 3 successfully purchased 6 books, and the current remaining stock is 18
  User 8 successfully purchased 3 books, and the current remaining stock is 11
  User 5 successfully purchased 9 books, and the current remaining stock is 1
  Stock is not enough, Stop!
  Stock is not enough, Stop!
  Stock is not enough, Stop!
  Stock is not enough, Stop!
  User 2 successfully purchased 4 books, and the current remaining stock is 6
  User 7 successfully purchased 1 books, and the current remaining stock is 0
  Stock is not enough, Stop!
  Stock is not enough, Stop!
  Stock is not enough, Stop!
  Stock is not enough, Stop!
  =======================================
  Final stock: -36
  Total orders: 377
  Total purchased books: 2036
  Process duration: 97.72 seconds.
  Test finished！
  =======================================
  ```

- 回到 [终端 1](ssh://monitor1-ssh#1)（`test-branch-1`），该方案执行耗时最长，但库存最终准确归零，说明此方案效果正确、如预期所示。

  ```output
  ...
  User 5 successfully purchased 6 books, and the current remaining stock is 11
  User 3 successfully purchased 5 books, and the current remaining stock is 6
  User 1 successfully purchased 3 books, and the current remaining stock is 3
  User 6 successfully purchased 2 books, and the current remaining stock is 1
  User 2 successfully purchased 1 books, and the current remaining stock is 0
  Stock is not enough, Stop!
  Stock is not enough, Stop!
  Stock is not enough, Stop!
  Stock is not enough, Stop!
  Stock is not enough, Stop!
  Stock is not enough, Stop!
  Stock is not enough, Stop!
  Stock is not enough, Stop!
  =======================================
  Final stock: 0
  Total orders: 363
  Total purchased books: 2000
  Process duration: 181.34 seconds.
  Test finished！
  =======================================
  ```

### 步骤 5. 进入下一个测试阶段

- 在此步骤中，你需要将本模块的成功测试结果用于后续模块的测试，无论测试是否成功，都需要保留数据。

![demo_branch_pic_05.png 50%](https://lab-static.pingcap.com/images/2025/11/27/1e3707c00fbf73158eddf3f321b37edf31a22ba2.png)

- 由于终端 1 的方案效果正确，建议以该分支继续后续测试，同时保留所有测试数据。在 TiDB Cloud 控制台进入 [集群页面](https://tidbcloud.com/project/clusters)，点击你的实验集群名称。如果你通过 `创建新集群` 开始实验，集群名为 `${LAB:SERVERLESS_CLUSTER_NAME}`。

![Demo-RU-001.png 85%](https://lab-static.pingcap.com/images/2025/11/24/9c016ddc77e89cb9d9ae0669890d36cb316f6d20.png)

- 点击集群名称 `${LAB:SERVERLESS_CLUSTER_NAME}`，然后在左上角选择 `Overview`（概览）。

![Demo-RU-002.png 85%](https://lab-static.pingcap.com/images/2025/11/24/8d064cdda27759df1640171cbf92f7d340e4364d.png)

- 点击左侧的 `Branches`（分支）标签，打开分支面板。

![demo_branch_UI_01.png 85%](https://lab-static.pingcap.com/images/2025/11/27/c094e9effcea9e16423f52268a3f03be1d9f27fe.png)

- 在分支面板右上方点击 `Create Branch`（创建分支）按钮，新建一个分支。

![demo_branch_UI_13.png 85%](https://lab-static.pingcap.com/images/2025/11/27/5009833f8220f1851e54711e7ce3eec3bb8b0e90.png)

- 在弹出的创建分支窗口，Name 字段填写 `test-branch-1-1`，Parent branch 选择 `test-branch-1`，Include data up to 选择 `Current point in time`（当前时点），即以 `test-branch-1` 当前数据快照创建名为 `test-branch-1-1` 的分支，并点击创建按钮。

![demo_branch_UI_14.png 50%](https://lab-static.pingcap.com/images/2025/11/27/ce6719b47458cca94d43ef005bde46298527f39d.png)

- 等待 `test-branch-1-1` 的状态变为 `Active`（活跃）。此分支用于后续测试，同时继续保留 `test-branch-1`、`test-branch-2` 和 `test-branch-3` 以便回归测试时使用。

![demo_branch_UI_15.png 85%](https://lab-static.pingcap.com/images/2025/11/27/4d3232d62da501d83d91d9277479b4ddf8dfa798.png)

### 步骤 6. 实验清理

- 在 TiDB Cloud 控制台进入 [集群页面](https://tidbcloud.com/project/clusters)，点击你的实验集群名称。如果你通过 `创建新集群` 开始实验，集群名为 `${LAB:SERVERLESS_CLUSTER_NAME}`。

![Demo-RU-001.png 85%](https://lab-static.pingcap.com/images/2025/11/24/9c016ddc77e89cb9d9ae0669890d36cb316f6d20.png)

- 点击集群名称 `${LAB:SERVERLESS_CLUSTER_NAME}`，然后在左上角选择 `Overview`（概览）。

![Demo-RU-002.png 85%](https://lab-static.pingcap.com/images/2025/11/24/8d064cdda27759df1640171cbf92f7d340e4364d.png)

- 点击左侧的 `Branches`（分支）标签，打开分支面板。

![demo_branch_UI_01.png 85%](https://lab-static.pingcap.com/images/2025/11/27/c094e9effcea9e16423f52268a3f03be1d9f27fe.png)

- 在分支面板中选择 `test-branch-1-1`，点击 `Action`（操作）栏的 `...` 菜单，选择 `Delete`（删除）。

![demo_branch_UI_16.png 85%](https://lab-static.pingcap.com/images/2025/11/27/bd50c1b3885fe2492471bc20d2005f327d7a20ad.png)

- 在弹出确认窗口中点击 `I understand, delete`（我已知晓，确认删除）。

![demo_branch_UI_17.png 50%](https://lab-static.pingcap.com/images/2025/11/27/72bedd61a6ce908307edb8e6950df88c56466ede.png)

- 按上述方式依次删除 `test-branch-2`、`test-branch-3` 以及 `test-branch-1`。

![demo_branch_UI_18.png 85%](https://lab-static.pingcap.com/images/2025/11/27/dc0a91b1c61ac243809ad40a31f73ffca2afa7d8.png)

> [!NOTE]
> 本模块中，你手动管理和探索这三个分支。在实验 [使用 TiDB Cloud Branching 构建 Agent 分支工作流](/labs/demo_411) 中，我们将演示如何借助 API 以及 AI Agent 实现自动化管理分支，从而让分支操作实现远超三个分支、规模化的并行处理。