深入Redis学习教程之缓存，事务和分布式锁

# coding=utf-8

import redis,pymysql,time

r = redis.Redis(host='127.0.0.1', port=6379)

def getValueByDb(id):
    pass    # 略

def getValueById(prefix, id):   # prefix是key前缀
    key = prefix + "_" + id
    value = r.get(key)

    # 缓存没有命中
    if not value:
        lock_name = "lock_" + key
        if not r.get(lock_name):    # 某用户第一次触发击穿缓存
            r.set(lock_name, 1)
    
            # 从mysql中读取数据
            value = getValueByDb()
            r.setex(key, 86400, value)
            r.delete(lock_name)        # 解锁
        else:   # 数据还在从mysql中查询中，阻塞住其他并发查询的用户
            time.sleep(0.05)    # 阻塞50毫秒再查
            value = getValueById(prefix, id)
            
    return value    
    

multi    # 开启事务    
# 一系列操作
exec    # 执行操作
discard # 在multi后执行，可以撤销事务中的操作

BEGIN：显式地开启一个事务；
COMMIT：提交事务，将对数据库进行的所有修改变成为永久性的；
ROLLBACK：结束用户的事务，并撤销正在进行的所有未提交的修改；

multi
set a 1
set b 2
aaa            # 报错导致队列中所有命令都discard
set c 3
exec

# 结果： a b c 都为null

multi
set a 1
set b 2
discard        # 删掉队列中discard前的命令
set c 3
exec

# 结果： a b 为null，c为3

multi
set a 1
set b name
incr b        # 会报错，redis会忽略这一条命令，不影响其他命令
set c 3
exec

# 结果： a 1, b name,c 3

set num 10
multi
incr num
# 此时有人在另一个客户端中先执行了incr num
exec

# incr的结果是12

begin;
update transaction set num=num+1 where id=1;
select * from transaction;        # 结果为11，说明他是直接执行，而不像redis中放在队列等之后再执行

#此时再开另一个mysql客户端，查询num字段，发现结果为10，说明这个修改还没有真正写到磁盘中。
commit;        # 写入到磁盘中。

begin;
update transaction set num=num+1 where id=1;
select * from transaction;        # 结果为11，说明他是直接执行，而不像redis中放在队列等之后再执行

#此时再开另一个mysql客户端，update num字段为num-1，发现光标在等待，命令的执行被阻塞住了
commit;        # 写入到磁盘中。此时另一个客户端的命令也执行了。

#最终的结果为num = 10

# money和goods的初始值为1000 和 0

watch money goods
multi
set money 500        # 此时不会马上执行，而是先将这个命令放入队列中
set goods 5            # 此时不会马上执行，而是先将这个命令放入队列中
(在此时客户端2执行set goods 10)
exec

# 此时money还是1000，goods为10，事务没有生效
# 当事务结束的时候，watch的监控就会失效，下一次再执行multi的时候，其他客户端修改事务中的key就不会打断事务，所以下一次开启事务之前还要再watch一次才行。

# money和goods的初始值为1000 和 0

# 客户端1                      | # 客户端2
watch money goods    # 1      |  watch money goods  # 2
multi                 # 3     |  multi               # 4
set money 100         # 5     |  set money 1000       # 6
set goods 10         # 7      |  set goods 100    # 8
exec                 # 10     |  exec            # 9    

# coding=utf-8

from threading import Thread,Semaphore
import redis,time,random

# 连接redis
r = redis.Redis(host='127.0.0.1', port=6379)

# 定义线程类
class MiaoShaThread(Thread):
    def __init__(self, redis, sem, thread_no):
        super(MiaoShaThread, self).__init__()
        self.r = redis
        self.sem = sem
        self.thread_no = thread_no      # 记录线程的序号

    def run(self):
        with self.r.pipeline() as pipe:     # 在python中watch必须配合pipeline使用
            try:
                # 监视商品库存 inventory
                pipe.watch('inventory')

                # 获取库存必须放在multi之前，否则取到的是一个pipeline对象而不是value值
                # 获取库存必须放在watch之后，否则可能出现一获取到库存之后，库存马上被别的线程改为0，然后本线程才执行watch的情况
                inventory = int(pipe.get('inventory'))

                # 判断商品库存 inventory 是否大于0
                if inventory > 0:
                    pipe.multi()  # 开启事务,开启事务不能放在if之外
                    pipe.decr('inventory')      # 商品数-1
                    pipe.execute()              # 执行事务
                    print("第%s号用户抢购成功" % str(self.thread_no))
                    # time.sleep(random.uniform(0,1))
                else:
                    print("第%s号用户抢购失败" % str(self.thread_no))
            except:     # 如果出现异常说明在执行事务的过程中，其他线程先对库存进行了修改导致本线程执行事务失败
                print("第%s号用户抢购失败，您的商品被别人抢走了" % str(self.thread_no))

        self.sem.release()

# 用15个线程模拟50000个人抢购商品，一直维持着有15个线程在工作
thread_list = []
thread_num = 15
sem = Semaphore(thread_num)
for i in range(5000):
    sem.acquire()
    thread = MiaoShaThread(r, sem, i+1)
    thread_list.append(thread)
    thread.start()

for thread in thread_list:
    thread.join()

print(r.get('inventory'))

# coding=utf-8

from threading import Thread, Semaphore
import redis,uuid
from time import sleep

class DistributedLock:
    def __init__(self, redis, name, expire=2, timeout=3):
        self.r = redis
        self.expire = expire      # 锁的过期时间
        self.timeout = timeout      # 等待获取锁的时间    
        self.name = "lock_" + name          # 锁的key名
        self.uuid = str(uuid.uuid4())    # 锁的唯一标识，不同客户端生成的锁的唯一标识不同

    # 上锁
    def lock(self):
        while True:
            if not self.r.get(self.name):   # 如果其他线程或进程没有拿到锁，则本线程就先持有锁
                self.r.set(self.name, self.uuid)
                return True
            else:   # 如果没有拿到锁，则等待其他线程释放锁
                print("锁被占用")
                sleep(0.01)

    # 解锁
    def unlock(self):
        self.r.delete(self.name)

class BuyGoods(Thread):
    def __init__(self, redis, sem, thread_no):
        super(BuyGoods, self).__init__()
        self.r = redis
        self.thread_no = thread_no
        self.sem = sem
        self.lock = DistributedLock(self.r, "goods_inventory")

    def run(self):
        self.lock.lock()
        if int(self.r.get("inventory")) > 0:
            self.r.decr("inventory")
            print("抢购成功:" + str(self.thread_no + 1))
        else:
            print("库存不足")
        self.lock.unlock()

        self.sem.release()


# 连接redis
r = redis.Redis(host='127.0.0.1', port=6379)

# 用15个线程模拟50000个人抢购商品，一直维持着有15个线程在工作
thread_list = []
thread_num = 15
sem = Semaphore(thread_num)
for i in range(5000):
    sem.acquire()
    thread = BuyGoods(r, sem, i)
    thread_list.append(thread)
    thread.start()

for thread in thread_list:
    thread.join()

print(r.get('inventory'))

if self.r.setnx(self.name, self.uuid):
    self.r.expire(self.name, self.expire)
else:
    sleep(0.01)