乐观锁和悲观锁配合事务的应用

Posted 2021-03-20 chanyuli

概念

乐观锁

总是假设最好的情况，认为竞争总是不存在，每次拿数据的时候都认为不会被修改，因此不会先上锁，在最后更新的时候比较数据有无更新，可通过版本号或CAS实现。

悲观锁

总是假设最坏的情况，认为竞争总是存在，每次拿数据的时候都认为会被修改，因此每次都会先上锁。其他线程阻塞等待释放锁。我们之前使用的线程锁之类的，都是悲观锁。

两种锁的使用场景

悲观锁：用于写比较多的情况，避免了乐观锁不断重试从而降低性能。具体理解为，悲观锁每次读数据就会上锁，让别的线程无法读。所以比较适合写数据操作比较多的情况。

乐观锁：用于读比较多的情况，避免了不必要的加锁的开销。具体看下面例子。

乐观锁应用场景

场景：用户下了订单之后，两个小时内如果不付款，后台自动把订单作废。

分析：用户已经下了订单了，所以库存会有减少，订单分成小订单，比如要买A两件和B三件，两小时后作废订单，并且让对应商品库存增加订单里的物品的数量。

看到两小时后才做操作，第一就要想到异步，这时候celery是非常符合需求的。使用celery来实现延时任务。

首先建一个文件夹，名字随意，文件夹下建一个名字为celery.py的文件。

在需要异步执行任务的方法里调用

from pro_celery.celery import del_order
from  datetime import datetime
def check_order(order_id,second=7200):
    #获取当前时间并计算出延迟执行的时间。
    ctime = datetime.now()
    utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())
    from datetime import timedelta
    time_delay = timedelta(seconds=second)
    task_time = utc_ctime + time_delay
    #提交任务，第一个参数为订单的id
    result = del_order.apply_async(args=[order_id, ], eta=task_time)

celery

import celery
import time

#连接你的redis数据库
# broker='redis://127.0.0.1:6379/2' 不加密码
backend = 'redis://127.0.0.1:6379/1'
broker = 'redis://127.0.0.1:6379/2'
cel = celery.Celery('test', backend=backend, broker=broker)

import os, sys
import django

BASE_DIR = os.path.dirname(os.path.dirname(__file__))  # 定位到你的django根目录
# sys.path.append(os.path.join(BASE_DIR, "app01"))
sys.path.append(os.path.abspath(BASE_DIR))
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "shop.settings")
django.setup()
from django.db import transaction


@cel.task
#事务
@transaction.atomic
#关键代码
def del_order(order_id):

    from app01 import models

    # 查看订单数据，查找传进来的id的未支付的订单
    order_data = models.Order.objects.filter(order_id=order_id, pay_status=False).first()

    # 如果有数据表示没有支付，要进行库存回滚，和取消订单
    if order_data:

        # 去Order_items表里获取该订单下的所有子订单
        order_items = models.Order_items.objects.
        filter(order_id=order_id).all()

        # 字典生成式将子订单中的数据转变成 {商品id:购买数量，。。。}的格式
        product_all_dic = {item.product_id: item.nums for item in                order_items}

        # 获取所有商品的id，成为list格式
        product_all_id = list(product_all_dic.keys())

        # 获取所有的商品
        all_product = models.Product.objects.filter(product_id__in=
        product_all_id).all()
        #在这个地方开启事务
        sid = transaction.savepoint()

        # 把对应的商品进行库存回滚
        for product in all_product:
            #循环三次就可以了，如果三次都还没有成功回滚，就重新执行这个异步任务
            for i in range(3):
                #这一步实际上是查表，跨表查询商品的库存
                stock = product.stock.quantity
                #回滚后的该商品的库存
                new_stock= stock+product_all_dic[product.product_id]

                #乐观锁，在这里的查询条件中有一个quantity=stock，判断在上面的跨表查询后，到现在库存有没有发生变化，没有的话就更新这个商品对应的库存（注意，这里循环的是每个商品，可能会出现一共三个商品需要回滚，在你回滚了两个，准备回滚第三个的时候，有人下单了，这时候下面的res就没值课，需要把前面的两个回滚全部作废，重新开始整个回滚），有的话就说明有人在操作数据库，不能够回滚，所以会进入下面的if里面。
                res = models.Stock.objects.filter(stock_id=
                product.stock.stock_id, quantity=stock).update(
                    quantity=new_stock)

                if not res:
                    #循环到了第三次了，还是没有res，说明这段时间都有人在操作数据库，显然再等不合理，于是直接准备开始下一次数据回滚。
                    if i == 2:
                        #这一步是事务回滚。把从上面的开启事务开始，到这里，对数据的操作全都作废，因为三个商品，只要有一个没改成功，就得全部作废。
                        transaction.savepoint_rollback(sid)

                        # 如果这个执行失败了，那我们要从新提交任务，不然库存无法回滚，也就是说，这个celery的任务，最后一定会成功。
                        from app01.common import func
                        func.check_order(order_id, 1)
                        return
                    #如果i不等于2，就直接执行下一次循环
                    else:
                        continue
                # res有值，走到这里说明一个商品的数据成功修改了
                else:
                    break
        # 修改订单状态
        #走到了这里，就代表所有的商品库存都改掉了，接下来只用修改订单状态，把订单都改成死订单就好了
        res1 = models.Order.objects.filter(order_id=order_id, pay_status
        =False).update(status="dead")
        if res1:
            #如果订单修改成功提交事务
            transaction.savepoint_commit(sid)
        else:
            #否则事务回滚
            transaction.savepoint_rollback(sid)