Flask单机测试，实现redis+MQ秒杀业务，防止超限

简单描述

抢购，秒杀是商场业务很常见的应用场景，主要需求解决：

1.高并发

2.如何解决库存的正确减少（"超卖"的问题）

redis 命令说明exists >setnx> incrby

# (1) 设置一些参数
    # 限制商品库存总数
    amount_limit = 1000
    # 设置redis中的缓存key name
    keyname = ‘limit‘
    # incrby 每次自动增加的数量我们设置为1
    incr_amount = 1
    # (2)判断redis中是否有我们的key name 也就是库存商品的建
　　　　通过r.exists()
　　

　　# 测试我们设置库存量初始值为0
　　# setnx >>>可以防止并发时多次设置key 
　　r.setnx(keyname, 0)  # 初始值0
　　# 判断自增加数和库存数
　　incr_num = r.incrby(keyname, incr_amount)

　测试代码

from flask import Flask
import logging
from logging import handlers
import redis

# 添加日志

rf_handler = handlers.TimedRotatingFileHandler(
    ‘redis.log‘, when=‘midnight‘, interval=1,
    backupCount=7)
rf_handler.setFormatter(
    logging.Formatter(
        "%(asctime)s %(filename)s line:%(lineno)d [%(levelname)s] %(message)s")
)

logging.getLogger().setLevel(logging.INFO)
logging.getLogger().addHandler(rf_handler)

app = Flask(__name__)

# 链接我们的redis connect redis
pool = redis.ConnectionPool(host=‘localhost‘, port=6379,
                            decode_responses=True)
r = redis.Redis(connection_pool=pool)


# 操作方法 进行商品库存的读取
# redis.get(store_shop)
# 为了简单的演练 我们自己自定义参数

def limit_handler():
    """
    :return True:允许 ；False:拒绝

    """
    # (1) 设置一些参数
    # 限制商品库存总数
    amount_limit = 1000
    # 设置redis中的缓存key name
    keyname = ‘limit‘
    # incrby 每次自动增加的数量我们设置为1
    incr_amount = 1
    # (2)判断redis中是否有我们的key name 也就是库存商品的建

    if not r.exists(keyname):
        """
        # 测试我们设置库存量初始值为0
        # setnx >>>可以防止并发时多次设置key  

        """
        r.setnx(keyname, 0)  # 初始值0

        # 数据插入redis 后进行判断是否大于限制数
    # 判断自增加数和库存数
    incr_num = r.incrby(keyname, incr_amount)

    if incr_num <= amount_limit:


        """
        再次将商品的信息进行缓存到redis中:user_id order_id price
        创建商品订单号
        def post(self, request, *args, **kwargs):
        # 1）获取前台信息：商品、价格、支付方式
        request_data = request.data
        subject = request_data.get(‘subject‘)
        total_amount = request_data.get(‘total_amount‘)
        pay_type = request_data.get(‘pay_type‘)
        if not (subject and total_amount and pay_type):
            return APIResponse(2, ‘数据有误‘)
        # 2）生成订单（订单号，订单表的订单记录）
        out_trade_no = str(time.time())
         try:
            user_obj = models.Order.objects.create(subject=subject, total_amount=total_amount, pay_type=pay_type,
                                        out_trade_no=out_trade_no, user=user)
            print(user_obj,‘order999‘)
         except:
            return APIResponse(1, ‘订单生成失败‘)
        # 3）生成支付链接，并返回
        order_string = alipay.api_alipay_trade_page_pay(
            out_trade_no=out_trade_no,
            total_amount=total_amount,
            subject=subject,
            return_url=settings.RETURN_URL,
            notify_url=settings.NOTIFY_URL
        )
        order_url = pay_url + order_string
        return APIResponse(order_url=order_url)
        """
        # out_trade_no
        # 条件有限
        import time
        out_trade_no = str(time.time())
        return True  # 正常是生成订单 >>>保存商品信息>>>调用alipay进行页面的跳转》》》MQ写入队列 进行数据的写入
    return False


@app.route(‘/home‘)
def Login():
    return ‘ok‘


@app.route(‘/kill‘)
def hot():

    if limit_handler():

        logging.info("successful")
        return ‘提交订单成功‘
    else:
        logging.info("failed")

        return "商品已售完"


if __name__ == ‘__main__‘:
    app.run(debug=True)