python,事务,多线程

Question

问题：多线程和事务

       @action(methods=['post'], detail=False)
    def uploadTest(self, request):
        # 目标：1.假如每次随机数都为3 ，则能完整插入所有数据，msgList返回空数组
        #      2.随机数出现有一次2/5的倍数，则希望统计出所有报错的随机数，并且数据库保持没有数据插入
        # 当前能实现1，也能实现2的返回所有错误情况，但2时，主线程保持了原子性，子线程没有
        data = {'code': 0, 'msg': '', 'data': []}
        msgList = []
        try:
            with transaction.atomic():
                # 主线程会有一次数据库操作
                connection.cursor().execute("INSERT INTO upload_main (id, name,age) VALUES (%s, %s, %s)", (666, 'name', 666))
                for i in range(0, 5):
                    rand = random.randint(1, 9)
                    print(rand)
                    tempList = pool.submit(uploadSun, rand , msgList)
                msgList = tempList.result()
                if len(msgList)>0:
                    raise Exception('error')
        except Exception as e:
            data["data"] = msgList
            return JsonResponse(data)
        data['msg']='success'
        return JsonResponse(data)

def uploadSun(random,msgList):
    # 子线程先判断,完成后进行第一次数据库操作
    if random % 2 == 0:
        # 模拟如果该子线程报错，则返回报错信息统计返回主线程
        msgList.append(random)
    else:
        connection.cursor().execute("INSERT INTO upload_not2 (id, name,age) VALUES (%s, %s, %s)", (random, 'name2', random))
    # 子线程完成后进行第二次数据库操作
    if random % 5 == 0:
        # 模拟如果该子线程报错，则返回报错信息统计返回主线程
        msgList.append(random)
    else:
        # 实际业务第二次是第一次的关联表,需要加上第一次插入后的id
        connection.cursor().execute("INSERT INTO upload_not5 (id, name,age) VALUES (%s, %s, %s)", (random, 'name2', random))
    return msgList

目前：

Answer 1

这段代码中涉及了多线程和事务，主要的问题在于子线程中的数据库操作无法参与到主线程的事务中，因此当子线程执行失败后，主线程中已经成功的数据库操作也无法回滚。

为了解决这个问题，可以将主线程和子线程中的数据库操作都放在同一个事务中，即在with transaction.atomic()块中处理主线程和子线程的数据库操作，这样无论哪个线程中的操作失败了都能进行回滚，代码如下：

from django.db import transaction

@action(methods=['post'], detail=False)
def uploadTest(self, request):
    data = {'code': 0, 'msg': '', 'data': []}
    msgList = []
    try:
        with transaction.atomic():
            # 主线程会有一次数据库操作
            connection.cursor().execute("INSERT INTO upload_main (id, name,age) VALUES (%s, %s, %s)", (666, 'name', 666))
            # 创建一个新的连接，用于子线程的数据库操作
            with connection.cursor() as cursor:
                for i in range(0, 5):
                    rand = random.randint(1, 9)
                    print(rand)
                    tempList = pool.submit(uploadSun, rand , msgList, cursor)
                msgList = tempList.result()
            if len(msgList) > 0:
                raise Exception('error')
    except Exception as e:
        data["data"] = msgList
        return JsonResponse(data)
    data['msg'] = 'success'
    return JsonResponse(data)
 
def uploadSun(random, msgList, cursor):
    # 子线程先判断,完成后进行第一次数据库操作
    if random % 2 == 0:
        # 模拟如果该子线程报错，则返回报错信息统计返回主线程
        msgList.append(random)
    else:
        cursor.execute("INSERT INTO upload_not2 (id, name,age) VALUES (%s, %s, %s)", (random, 'name2', random))
    # 子线程完成后进行第二次数据库操作
    if random % 5 == 0:
        # 模拟如果该子线程报错，则返回报错信息统计返回主线程
        msgList.append(random)
    else:
        # 实际业务第二次是第一次的关联表,需要加上第一次插入后的id
        cursor.execute("INSERT INTO upload_not5 (id, name,age) VALUES (%s, %s, %s)", (random, 'name2', random))
    return msgList

在这个新的代码中，我们首先打开了一个新的连接cursor，用于子线程的数据库操作，然后将这个连接作为参数传递给uploadSun函数。在主线程中，我们将整个处理流程放在with transaction.atomic()块中，包括主线程和子线程的数据库操作。这样，当任何一个线程中的数据库操作失败时，整个事务都会回滚，从而保证了数据的一致性。