微服务之分布式锁

悲观锁和乐观锁

• 悲观锁：一开始就认为着里会出现锁的竞争，给自己加一把锁

  ​ 比如一些人为了防止犯错误，一开始就要等到时机非常成熟的时候才会行动，比如锁一样，他拿到一把锁之后才去执行，抱着一种悲观的态度。

• 乐观锁：不关心执行的时出现错误，如果出现问题我重新执行一次，不停的尝试

  ​ 有一些人呢，他就是愿意去尝试，什么都没准备好就开始做了，失败了我重新尝试，抱着一种乐观的态度。

基于 Mysql 乐观锁

• 优点

  ​ 简单

  ​ 不需要额外的组件 - 维护，mysql的维护比较简单 - 最合适的才是最好的。 系统的可用性

• 缺点

  ​ 性能

import threading
from peewee import *
from inventory_srv.config import config


class Inventory(Model):
    # 商品的库存表
    goods = IntegerField(verbose_name="商品id", unique=True)
    stocks = IntegerField(verbose_name="库存数量", default=0)
    version = IntegerField(verbose_name="版本号", default=0)  # 分布式锁的乐观锁
    class Meta:
        database = config.DB

def sell():
    # 多线程下的并发带来的数据不一致的问题
    goods_list = [(1, 10), (2, 20), (3, 30)]
    with config.DB.atomic() as txn:
        # 超卖
        for goods_id, num in goods_list:
            # 查询库存 如果更新失败增重复尝试
            while True:
                goods_inv = Inventory.get(Inventory.id== goods_id)
                print(f"当前的版本号：{goods_inv.version}")
                print(f"商品{goods_id} 售出 {num}件")
                import time
                from random import randint

                time.sleep(randint(1, 3))
                if goods_inv.stocks < num:
                    print(f"商品：{goods_id} 库存不足")
                    txn.rollback()
                    break
                else:
                    # 让数据库根据自己当前的值更新数据， 这个语句能不能处理并发的问题
                    # 我当时查询数据的时候版本号是 goods_inv.version
                    query = Inventory.update(stocks=Inventory.stocks - num, version=Inventory.version + 1).where(
                        Inventory.id goods_id, Inventory.version == goods_inv.version)
                    ok = query.execute()
                    if ok:
                        print("更新成功")
                        break
                    else:
                        print("更新失败")


    
if __name__ == "__main__":
    config.DB.create_tables([Inventory])
 	t1 = threading.Thread(target=sell)
    t2 = threading.Thread(target=sell)
    t1.start()
    t2.start()

    t1.join()
    t2.join()
    # 演示基于数据库的乐观锁机制
  

基于 Redis 分布式锁

Python Redis驱动: https://github.com/redis/redis-py

Python

基本的 Redis 分布式锁

当并发很高的时候使用 get();set()可能会存在同时获取数据的问题，一定要使用 setnx 来保证数据的原子性。

import threading
from peewee import *
from inventory_srv.config import config, server_config
import redis

# 库存表定义
class Inventory(Model):
    # 商品的库存表
    goods = IntegerField(verbose_name="商品id", unique=True)
    stocks = IntegerField(verbose_name="库存数量", default=0)
    version = IntegerField(verbose_name="版本号", default=0)  # 分布式锁的乐观锁
	class Meta:
        database = config.DB


class RedisLock:
    def __init__(self, key, uuid=None):
        if uuid is None:
            import uuid
            uuid = uuid.uuid4()
        self.id = uuid
        self.redis_client = redis.Redis(host=server_config.REDIS_HOST, port=server_config.REDIS_PORT)
        self.key = f"{server_config.REDIS_PREFIX['inventory']}{key}"

    def acquire(self):

        while True:
            # if not self.redis_client.get(self.key):  # 错误的示范，高并发时可能出现同时进入代码的情况，不能保证原子性
            #     self.redis_client.set(self.key, 1)
            #     break
            # 获取到锁
            # 使用 setnx 获取不到时插入数据，保证数据的原子性
            # if self.redis_client.setnx(self.key, 1):  # 如果 key 存在返回False，不存在设置 key 并返回 True
            if self.redis_client.set(self.key, self.id, nx=True, ex=15):

                # 启动一个线程去定时刷新过期时间 最好也使用lua脚本完成 - 看门狗
                break
            else:  # 获取不到锁， 等待一秒重新获取
                import time
                time.sleep(1)

    def release(self):
        # 防止自己的锁被别人删除
        # 这里的存在文件，需要将代码原子化 redis 没有提供这样的接口，我们可以使用 lua 脚本让redis原子化执行
        id = self.redis_client.get(self.key)
        print(f"释放{self.key}")
        if id is not self.id:
            print("不能删除不属于自己的出锁")
            return
        self.redis_client.delete(self.key)



def sell():
    # 多线程下的并发带来的数据不一致的问题
    goods_list = [(1, 10), (2, 20), (3, 30)]
    with config.DB.atomic() as txn:
        # 超卖
        for goods_id, num in goods_list:
            lock = RedisLock(key=goods_id)
            # 查询库存 如果更新失败增重复尝试
            goods_inv = Inventory.get(Inventory.id == goods_id)
            print(f"商品{goods_id} 售出 {num}件")
            import time
            from random import randint

            time.sleep(randint(1, 2))
            if goods_inv.stocks < num:
                print(f"商品：{goods_id} 库存不足")
                txn.rollback()
                break
            else:
                # 获取锁
                lock.acquire()
                # 让数据库根据自己当前的值更新数据，
                query = Inventory.update(stocks=Inventory.stocks - num).where(Inventory.id == goods_id)
                ok = query.execute()
                if ok:
                    print(f"商品：{goods_id} 更新成功")
                else:
                    print(f"商品：{goods_id} 更新失败")
                # 释放锁
                lock.release()


if __name__ == "__main__":
    config.DB.create_tables([Inventory, InventoryHistory])

    t1 = threading.Thread(target=sell)
    t2 = threading.Thread(target=sell)
    t3 = threading.Thread(target=sell)
    t4 = threading.Thread(target=sell)
    t5 = threading.Thread(target=sell)
    t1.start()
    t2.start()
    t3.start()
    t4.start()
    t5.start()

    t1.join()
    t2.join()
    t3.join()
    t4.join()
    t5.join()

python-redis-lock

https://github.com/ionelmc/python-redis-lock

使用pip 安装或着拷贝代码
https://github.com/ionelmc/python-redis-lock/blob/master/src/redis_lock/__init__.py

比较简单就不演示了