Django transaction.atomic 事務的使用

函數 transaction.atomic

數據庫的讀寫操作中，事務在保證數據的安全性和一致性方面起着關鍵的作用，而回滾正是這里面的核心操作。

遇到並發的時候常常會因為接口的訪問順序或者其他情況，導致數據的不一致，這時候就知道回滾的重要性了，下面有個簡單的小例子 可以實現數據庫事務的操作

Demo

數據庫的模型和返回值窩這邊就不貼出來了
這是一個接口

from django.db import transaction
import time
def keke_test(request):
    code = 1
    msg = '操作成功'

    g = request.REQUEST.post
    params = dict()
    model_id = g.get('id', '0')
    model = None
    if model_id > 0:
        model = IosSdkVersions.objects.filter(id=model_id).first()
        with transaction.atomic():  # 禁止自動提交,保證該函數中的所有數據庫操作在同一個事物中，第一個數據庫操作1即使成功保存到數據庫中，只要第2個數據操作失敗，那么所有該段代碼所有設計的都會更改回滾到原來
            try:
                sdk_flag_model = IosSdkTest.objects.filter(id=1).first()
                sid = transaction.savepoint()  # 開啟事務設置事務保存點
                sdk_flag_model.sdk_flag = 8
                sdk_flag_model.save() # 完成就保存，但不是馬上提交
                time.sleep(10)
                all_issue_field_value = IosSdkVersions.objects.filter(id=3)
                for issue in all_issue_field_value:
                    issue.test_id = 8
                    issue.save()
                    # 1 / 0
            except Exception as e:
                print(e)
                transaction.savepoint_rollback(sid)  # 失敗回滾事務(如果數據庫操作發生異常，回滾到設置的事務保存點)
                print('數據保存失敗')
                code = -1001
            else:
                code = 1
                transaction.savepoint_commit(sid)  # 如果沒有異常，成功提交事物

    params['code'] = code
    return json_response(params)

# 接口2，當執行第一個接口的時候，因為sleep的存在，這個操作會一直等待到第一個接口完成操作 再進行操作
def keke_test2(request):
    code = 1
    msg = '操作成功'
    _g = request.REQUEST.get
    sdk_flag_model = IosSdkTest.objects.filter(id=1).first()
    sdk_flag_model.sdk_flag = 2
    sdk_flag_model.save() # 因為被鎖住了所以接口會等到上一個接口的完成
    # if sdk_flag_model.sdk_flag == 8:
        # all_issue_field_value = IosSdkVersions.objects.filter(id=3)
        # for issue in all_issue_field_value:
        #     issue.test_id -= 2
        #     issue.save()
    params = dict()
    return json_response(params)

# 接口2 other
def keke_test2(request):
    code = 1
    msg = '操作成功'
    _g = request.REQUEST.get
    sdk_flag_model = IosSdkTest.objects.filter(id=1).first()
    if not sdk_flag_model.sdk_flag == 2: # 會等到flag為2的時候才去執行下面的操作
        sdk_flag_model.sdk_flag = 3
        sdk_flag_model.save()
    params = dict()
    params['code'] = code
    return json_response(params)

# other

def keke_test2(request):
    code = 1
    msg = '操作成功'
    _g = request.REQUEST.get
    with transaction.atomic():  # 禁止自動提交,保證該函數中的所有數據庫操作在同一個事物中，第一個數據庫操作1即使成功保存到數據庫中，只要第2個數據操作失敗，那么所有該段代碼所有設計的都會更改回滾到原來
        sid = transaction.savepoint()  # 開啟事務設置事務保存點
        try:
            sdk_flag_model = IosSdkTest.objects.select_for_update().get(id=1) # 互扯鎖的另一種實現
            sdk_flag_model.sdk_flag = 8
            sdk_flag_model.save()
            time.sleep(10)
        except Exception as e:
            print(e)
            transaction.savepoint_rollback(sid)  # 失敗回滾事務(如果數據庫操作發生異常，回滾到設置的事務保存點)
            print('數據保存失敗')
            code = -1001
        else:
            code = 1
            transaction.savepoint_commit(sid)  # 如果沒有異常，成功提交事物
    params = dict()
    params['code'] = code
    return json_response(params)

悲觀鎖

# 悲觀鎖實現， 都要放在 with transaction.atomic() 事務中 同條數據上鎖完，需要等釋放后才能再次上鎖的吧

def keke_test2(request):
    code = 1
    msg = '操作成功'
    _g = request.REQUEST.get
    with transaction.atomic():  # 禁止自動提交,保證該函數中的所有數據庫操作在同一個事物中，第一個數據庫操作1即使成功保存到數據庫中，只要第2個數據操作失敗，那么所有該段代碼所有設計的都會更改回滾到原來
        sid = transaction.savepoint()  # 開啟事務設置事務保存點
        try:
            sdk_flag_model = IosSdkTest.objects.select_for_update().get(id=1)  # 互扯鎖的另一種實現
            sdk_flag_model.sdk_flag = 8
            sdk_flag_model.save()
            time.sleep(10)
        except Exception as e:
            print(e)
            transaction.savepoint_rollback(sid)  # 失敗回滾事務(如果數據庫操作發生異常，回滾到設置的事務保存點)
            print('數據保存失敗')
            code = -1001
        else:
            code = 1
            transaction.savepoint_commit(sid)  # 如果沒有異常，成功提交事物
    params = dict()
    params['code'] = code
    return json_response(params)


def keke_test3(request):
    code = 1
    msg = '操作成功'
    _g = request.REQUEST.get
    with transaction.atomic():
        sdk_flag_model = IosSdkTest.objects.select_for_update().get(id=1)
        print(sdk_flag_model.sdk_flag)
    params = dict()
    params['code'] = code
    return json_response(params)

原生sql 鎖的操作

參考：https://segon.cn/mysql-for-update.html