Django+MySQL+Redis构建网上商城：订单状态机与支付宝集成实践

简介：这是一份基于Django+MySQL+Redis的Python网上商城完整源代码，面向具备一定Python基础、希望学习电商系统前后端开发或毕业设计参考的开发者。资源实现了从用户注册登录、商品展示与详情、关键词搜索，到按尺寸颜色数量加入购物车、购物车与地址管理、订单生成，再到支付宝支付接入的完整电商闭环，可直接作为二次开发或学习模板。包内共446个文件，以Python源码、HTML/CSS/JS前端页面、图片素材为核心，辅以配置文件与说明文档，压缩包约35.28MB，目录结构清晰，便于按功能模块查阅。目前已有1909人学习下载，适合用于理解Django项目分层、MySQL与Redis缓存配合，以及第三方支付接口的接入思路；源码中预留了支付宝公钥、应用私钥和appid的配置位置，并给出数据库连接修改提示，可帮助读者快速跑通项目并理解真实商城系统的核心链路。

1. Django 网上商城：先把订单状态机想清楚，再谈付款

真正让一个基于 Django+MySQL+Redis 的网上商城线上出事的，往往不是商品查询慢，而是订单状态机写得太随意。订单、库存、支付回调三个点一旦耦合，每次改需求都要两头补，退款时更会发现对不上账。这套技术栈本质是一条数据流水线：MySQL 存商品与订单事实，Redis 扛购物车、热榜和秒杀库存，支付宝付款负责把外部支付结果转成订单状态。

这个组合是小型自营商城和外包项目里出现频率最高的搭配。Django 的 ORM 和 Admin 能快速搭出商品后台，MySQL 的 InnoDB 事务能保证扣库存不超卖，Redis 把高频临时状态从数据库摘走，支付宝付款则省掉自建资金通道的合规与对账成本。下面按建表、缓存、支付、上线检查的顺序展开，每一步都给最小可运行代码，方便直接照着改。

2. 网上商城的 Django 数据模型与 MySQL 落库

2.1 商品、SKU 与订单：先画实体关系再写 models

一个商城最少要覆盖四类对象：SPU、SKU、订单、订单项。最常见的坑是把规格写死在一个 JSONField 里，后面要按颜色和尺寸查库存时，SQL 根本没法走索引；另一个坑是订单金额用 FloatField，浮点误差在退款时会变成 0.009 这种说不清的数字。我一般先把 SPU 和 SKU 拆开，再把订单与订单项做成主从表。

from django.db import models from django.core.validators import MinValueValidator class Category(models.Model): name = models.CharField(max_length=64, unique=True) parent = models.ForeignKey( "self", null=True, blank=True, on_delete=models.PROTECT, related_name="children" ) class Spu(models.Model): """商品主体，描述性字段放这里，规格不放""" title = models.CharField(max_length=128) category = models.ForeignKey(Category, on_delete=models.PROTECT) status = models.SmallIntegerField(default=1, help_text="1上架 0下架") class Sku(models.Model): """可下单的最小库存单位""" spu = models.ForeignKey(Spu, on_delete=models.CASCADE, related_name="skus") code = models.CharField(max_length=32, unique=True, help_text="商家编码") attrs = models.CharField(max_length=255, help_text="比如 颜色:红,尺码:L") price = models.DecimalField(max_digits=10, decimal_places=2) stock = models.PositiveIntegerField(default=0) class Order(models.Model): class Status(models.TextChoices): CREATED = "created", "已创建未支付" PAID = "paid", "已支付" SHIPPED = "shipped", "已发货" FINISHED = "finished", "已完成" CLOSED = "closed", "已关闭" order_no = models.CharField(max_length=32, unique=True) user_id = models.IntegerField(db_index=True) total_amount = models.DecimalField(max_digits=10, decimal_places=2) status = models.CharField( max_length=16, choices=Status.choices, default=Status.CREATED, db_index=True ) created_at = models.DateTimeField(auto_now_add=True) paid_at = models.DateTimeField(null=True, blank=True) class OrderItem(models.Model): order = models.ForeignKey( Order, on_delete=models.PROTECT, related_name="items" ) sku = models.ForeignKey(Sku, on_delete=models.PROTECT) sku_code = models.CharField(max_length=32) sku_attrs = models.CharField(max_length=255) price = models.DecimalField(max_digits=10, decimal_places=2) qty = models.PositiveIntegerField(validators=[MinValueValidator(1)])

这里有两个必调参数要说明。第一，金额一律用DecimalField，两位小数，下单时代码里统一用Decimal(str(price))计算，不要让前端直接把 float 传回来。第二，Status.TextChoices里的值本身是要落库的字符串，前端展示文案用get_status_display()拿，不要再来一个映射表。

OrderItem里冗余了sku_code和sku_attrs，这是刻意为之。订单生成后 SKU 可能改名或调价，订单明细必须保留下单瞬间的凭据，查询时也少一次 JOIN。on_delete=models.PROTECT的意义是禁止直接删商品和 SKU，避免历史订单指向不存在的记录。

2.2 用 Django ORM 扣库存：事务与 select_for_update 参数

秒杀和高并发加购场景下，扣库存必须防超卖。Django ORM 的F()表达式可以做原子更新，但下单还涉及创建订单、订单项、扣库存三个动作，必须包在同一个事务里。用select_for_update()给 SKU 行加排他锁，是 MySQL 侧最直观的兜底方案。

from decimal import Decimal from django.db import transaction from django.core.exceptions import ValidationError from .models import Order, OrderItem, Sku @transaction.atomic def create_order(user_id, cart_items): order = Order.objects.create( order_no=generate_order_no(user_id), user_id=user_id, total_amount=Decimal("0") ) total = Decimal("0") for item in cart_items: sku = Sku.objects.select_for_update().get(pk=item["sku_id"]) if sku.stock < item["qty"]: raise ValidationError(f"库存不足: {sku.code}") sku.stock -= item["qty"] # update_fields 只提交变化字段，减少一次全字段 UPDATE sku.save(update_fields=["stock"]) OrderItem.objects.create( order=order, sku=sku, sku_code=sku.code, sku_attrs=sku.attrs, price=sku.price, qty=item["qty"], ) total += sku.price * item["qty"] order.total_amount = total order.save(update_fields=["total_amount"]) return order

select_for_update().get(...)会对命中的 InnoDB 行加X 锁，直到当前事务提交或回滚。顺序拿锁可以显著降低死锁概率：所有线程都按pk升序锁 SKU 行，避免 A 订单先锁 sku1 再锁 sku2，B 订单反过来。update_fields参数走的是 Django 的预编译 UPDATE，只更新指定列，比全字段保存更快，也避免误提交并发改写过的其他字段。

@transaction.atomic必须直接放在最外层函数上。select_for_update只有在事务内才生效，如果被拆成内部函数并在外面另开事务，锁的细粒度会变得不可控。遇到OperationalError: Deadlock found时，先确认是否所有下单路径都按同一字段排序，再考虑把重试逻辑放在更高一层。

2.3 建库与索引检查：MySQL 侧把基线打好

业务代码写完后，先在 MySQL 建库和账号。注意字符集选utf8mb4，否则商品标题里的特殊符号会存不进去；排序规则选utf8mb4_unicode_ci在多数中文商城场景下够用。

mysql -uroot -p CREATE DATABASE mall DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; CREATE USER 'mall'@'localhost' IDENTIFIED BY 'StrongPassword_2024'; GRANT ALL PRIVILEGES ON mall.* TO 'mall'@'localhost'; FLUSH PRIVILEGES;

再把 Django 的 DATABASES 配置切过去。CONN_MAX_AGE要显式设置，Python 侧默认每次请求都新建 MySQL 连接，商城接口一多就会因为建连频繁把 CPU 耗在握手和认证上。

DATABASES = { "default": { "ENGINE": "django.db.backends.mysql", "NAME": "mall", "USER": "mall", "PASSWORD": "StrongPassword_2024", "HOST": "127.0.0.1", "PORT": "3306", "CONN_MAX_AGE": 60, "OPTIONS": { "charset": "utf8mb4", "init_command": "SET sql_mode='STRICT_TRANS_TABLES'", }, } }

跑完python manage.py makemigrations && python manage.py migrate后，用下面三条 SQL 核对索引基线。索引不是越多越好，更新频繁的 stock 字段就不要建单列索引，而订单表按(status, paid_at)联合索引能直接支持“待付款列表”和“超时关单”两类查询。

提示：MySQL 8 里SET sql_mode='STRICT_TRANS_TABLES'会拦截过长的字符串和非法日期，避免脏数据悄悄落库。支付宝回调或后台导入商品时常见的Data too long错误，多半就是没开严格模式。

3. 把 Redis 用进网上商城：购物车、热榜与秒杀库存

3.1 购物车用 Hash 而非 String，省掉一次反序列化

不少人把购物车整个 list 序列化成 JSON 塞进 String 型 key，用户每次点“加购”都要读出来、改、再写回去。并发双击加购时后一个写覆盖前一个，商品数量直接消失。GitHub 上不少类似“网上商城源代码”就是这个写法，代码看着短，压测一上去就掉链子。改用 Hash，每个字段存一个 SKU 的数量，hincrby是原子操作。

from django_redis import get_redis_connection r = get_redis_connection("default") def add_to_cart(session_key, sku_id, qty): key = f"cart:{session_key}" new_qty = r.hincrby(key, sku_id, qty) if int(new_qty) <= 0: r.hdel(key, sku_id) return new_qty def get_cart(session_key): key = f"cart:{session_key}" raw = r.hgetall(key) return {int(sku_id): int(count) for sku_id, count in raw.items()}

f"cart:{session_key}"这里的session_key对未登录用户来说通常是会话 ID，登录后把临时购物车合并到cart:user:{user_id}，然后删掉匿名 key。数量减到 0 或负数时直接hdel删除字段，而不是把负数继续留着。Hash 类型在 field 很少时是 ziplist 编码，内存占用比 String 更可控；SKU ID 是整数时还能用hstrlen快速判断购物车里有多少种商品。

3.2 商品热榜用 ZSET 加日期窗口

首页热榜最适合用 Redis 的 Sorted Set。每次用户浏览 SKU 详情页就zincrby一次，取榜单时zrevrange直接拿前 N 名。注意 key 里带日期，避免expire不断刷新导致榜单一个月都不重置。日榜 key 不带过期时间而用定时任务清理，也是常见做法，但停机或忘记清任务时 Redis 会堆积无用 key。

from django.conf import settings from django_redis import get_redis_connection r = get_redis_connection("default") def record_hit(sku_id, delta=1): # 按日分桶，第二天自动切新 key，旧 key 可交给 expire 回收 key = f"hot:skus:{timezone.now():%Y%m%d}" r.zincrby(key, delta, sku_id) if r.ttl(key) == -1: r.expire(key, 86400 * 2) def top_hot(limit=20): key = f"hot:skus:{timezone.now():%Y%m%d}" items = r.zrevrange(key, 0, limit - 1, withscores=True) return [(int(sku_id), int(score)) for sku_id, score in items]

zrevrange的withscores=True返回 float 分数，转 int 是为了避免把权重直接暴露给前端。按日分桶后，凌晨 0 点的榜单会空掉，如果业务不能接受，就改成“近 7 天滑动窗口”，用聚合任务每小时把 7 个日桶合并一次，再写回hot:skus:7d。分桶粒度越长，聚合频率越低，但榜单时效性越差。

3.3 秒杀预扣：Redis 库存与 Lua 原子性

秒杀活动里 MySQL 行锁会成为单点瓶颈。常见做法是活动开始前把可售库存预热进 Redis，扣减走 Lua 脚本保证“检查剩余量与扣减”两步原子执行。Lua 脚本在 Redis 中是串行执行的，不用考虑并发覆盖，脚本一次发给 Redis，返回结果再决定是否写 MySQL 订单。

-- KEYS[1] = seckill:1001:stock -- ARGV[1] = 本次扣减数量 local stock = tonumber(redis.call('get', KEYS[1]) or -1) if stock == -1 then return -2 -- 库存未初始化，可能活动没开始 end if stock >= tonumber(ARGV[1]) then redis.call('decrby', KEYS[1], ARGV[1]) return 1 -- 扣减成功 end return 0 -- 库存不足

Python 侧调用：

from django_redis import get_redis_connection r = get_redis_connection("seckill") LUA_SCRIPT = """ local stock = tonumber(redis.call('get', KEYS[1]) or -1) if stock == -1 then return -2 end if stock >= tonumber(ARGV[1]) then redis.call('decrby', KEYS[1], ARGV[1]) return 1 end return 0 """ def try_seckill(sku_id, qty): script = r.register_script(LUA_SCRIPT) return script(keys=[f"seckill:{sku_id}:stock"], args=[qty])

KEYS[1]和ARGV[1]的位置不能写反。register_script会把脚本缓存到 Redis，后续调用只传 key 和参数，省去每次传输脚本体的开销。扣减成功后的 MySQL 订单创建必须做“限流”和“一人一单”校验，Redis 库存只能挡超卖，挡不住脚本机器人。活动结束要把 Redis 里剩余的库存回写 MySQL，并记录实际成交量。

3.4 缓存穿透与击穿兜底

商品详情页和价格查询是商城缓存使用最频繁的地方。商品不存在时，一个不存在的 SKU 会被所有请求打到 MySQL，这就是穿透；某个热点 SKU 缓存过期瞬间，大量请求同时回源，这就是击穿。击穿最简单的兜底是“互斥锁重建缓存”。

import time CACHE_PREFIX = "sku:detail:" MUTEX_PREFIX = "lock:sku:detail:" def get_sku_detail(sku_id): cached = r.get(f"{CACHE_PREFIX}{sku_id}") if cached: return cached # 只让一个请求回源，其他请求短暂等待后重试 lock_key = f"{MUTEX_PREFIX}{sku_id}" if r.set(lock_key, "1", nx=True, ex=3): try: sku = Sku.objects.select_related("spu").get(pk=sku_id) r.set(f"{CACHE_PREFIX}{sku_id}", sku_to_json(sku), ex=60) return sku_to_json(sku) finally: r.delete(lock_key) time.sleep(0.05) return get_sku_detail(sku_id)

set(lock_key, "1", nx=True, ex=3)里的nx=True表示只有 key 不存在才能写入，ex=3是锁的过期时间，防止持有锁的线程在构建缓存时崩溃。缓存过期时间不要用固定 60 秒，建议加一个随机抖动，比如random.randint(30, 90)，避免整点大促时大量 key 同时失效。time.sleep(0.05)后递归重试是最偷懒的策略，生产上更稳的做法是用while循环加最大重试次数，防止异常时无限递归。

4. 支付宝付款的完整接入：从下单到异步通知闭环

4.1 接入支付宝前需要准备的密钥与配置

支付宝付款接入的核心不是前端跳转，而是“应用私钥签名，支付宝公钥验签”这个非对称流程。开放平台创建网页应用后，拿到 APPID，再自己生成一对 RSA2 密钥：应用私钥留在服务端签名，应用公钥上传给支付宝；支付宝平台会回传一把支付宝公钥，用来验证异步通知里的签名。两把公钥很容易弄混，我把它们放在一起对比。

沙箱环境会提供独立的 APPID 和密钥，调试时优先用沙箱，不要拿正式密钥在测试环境打真实流水。密钥建议只读一次后放入环境变量，不要提交到 Git 仓库。Django 的settings.py里用os.environ.get("ALIPAY_APP_ID")读取，配合.env文件管理，避免把私钥写进源码。

4.2 发起支付：构造 page.pay 参数与跳转收银台

用户在前端点击“去支付”时，后端不要直接把支付宝 SDK 对象暴露给模板，而是生成一个orderStr，再把它拼接成跳转 URL，前端window.location.href即可。官方支付网关地址由 SDK 默认值提供，不要手工拼接域名，防止环境切换后地址不一致。

import base64 from datetime import datetime def build_pay_url(order): alipay = get_alipay_client() # 入参必须是字符串，金额不能是 float order_string = alipay.api_alipay_trade_page_pay( out_trade_no=order.order_no, total_amount=str(order.total_amount), subject=f"网上商城订单-{order.order_no}", timeout_express="15m", # 订单支付超时时间 notify_url=settings.ALIPAY_NOTIFY_URL, # 异步回调 return_url=settings.ALIPAY_RETURN_URL, # 同步跳转，仅展示 ) return order_string

out_trade_no就是我们创建的order.order_no，长度通常限制在 64 位以内，项目中用时间戳加随机数生成即可。total_amount必须转成字符串，支付宝侧按金额精度校验，传 float 可能被序列化成带多位小数的科学计数法。timeout_express="15m"让支付宝在用户不支付时自动关闭交易，和 Django 侧定时关单任务配合，状态不会两边不一致。

4.3 异步通知验签与订单状态机

支付宝支付成功后，会主动 POST 到notify_url，字段包括out_trade_no、trade_no、trade_status、total_amount、sign。我的处理方式是：先验签，再查订单，再改状态，最后必须同步返回字符串success，否则支付宝会按递增间隔重复推送。

import logging from django.http import JsonResponse from django.views.decorators.csrf import csrf_exempt from django.utils import timezone logger = logging.getLogger(__name__) @csrf_exempt def alipay_notify(request): if request.method != "POST": return JsonResponse({"code": "error"}, status=400) data = request.POST.dict() signature = data.pop("sign", "") alipay = get_alipay_client() if not alipay.verify(data, signature): logger.warning("支付宝通知验签失败: %s", data.get("out_trade_no")) return JsonResponse({"code": "fail"}, status=400) order_no = data.get("out_trade_no") trade_status = data.get("trade_status") if trade_status == "TRADE_SUCCESS": with transaction.atomic(): order = Order.objects.select_for_update().get(order_no=order_no) if order.status == Order.Status.CREATED: order.status = Order.Status.PAID order.trade_no = data.get("trade_no") order.paid_at = timezone.now() order.save(update_fields=["status", "trade_no", "paid_at"]) # 返回 success 给支付宝，停止推送 return JsonResponse({"code": "success"})

trade_status只认TRADE_SUCCESS，不要用TRADE_FINISHED作为唯一支付成功的判断，部分新对接文档里TRADE_FINISHED表示交易完成后不再退款。select_for_update()锁住订单行，是为了防止“异步通知”和“用户手动刷新查询接口”同时进来时产生竞态。回调里再次校验total_amount与订单金额是否一致是必须的，否则订单金额被篡改而回调验签通过后会直接标记已支付。

4.4 对账与退款：把支付资源变成可审计的 MySQL 记录

异步通知只更新订单状态还不够，还要落一张支付流水表。支付宝的buyer_logon_id、invoice_amount和fund_bill_list等字段只有异步通知能拿到，很多商城项目事后审计退款时发现找不到当时的支付账号，就是这个表没建全。新表字段建议覆盖这几列：auto_no、trade_no、order_no、buyer_logon_id、total_amount、receipt_amount、payer_pay_amount、trade_status、raw_data。raw_data字段直接存回调原始 JSON，最关键的信息都不丢。

退款是独立的alipay.trade.refund接口，签名字段与支付不同，包含out_trade_no、refund_amount、out_request_no。out_request_no是退款请求号，同一个订单多次部分退款时靠它做幂等，必须每次生成新值。退款结果也走异步通知，要像支付回调一样落表，不要在前端 await 退款结果就关页面。

5. 上线前要给 Django 网上商城补的几刀：迁移、日志与压测

5.1 用 check --deploy 和日志配置兜住低级事故

python manage.py check --deploy --fail-level WARNING会检查 DEBUG、SECRET_KEY、ALLOWED_HOSTS、安全头等几十项配置，输出结果里每一项都值得过一遍。特别是SECRET_KEY，商城项目里很多源码包或教程直接把它写死在 settings.py，上线前必须换掉，否则你签名的 session 和支付密钥一旦被猜出，任何加密都没意义。

日志要提前配。异步通知验签失败、回调金额不一致、Redis 连接异常这三类日志必须单独打到文件。给logger.warning加上订单号和trade_no，事后查对账不用翻整个 request 日志。

5.2 给订单接口做幂等并跑一次并发压测

下单接口需要幂等键。前端点击“提交订单”后生成一个随机字符串放到请求头，后端查到这个键已存在就直接返回原有订单，而不是再创建一个新订单。这能挡掉用户双击和支付页面回退重提。接着用并发脚本压库存接口，验证超卖是否真的被挡在数据库外面。

from concurrent.futures import ThreadPoolExecutor, as_completed import requests ORDER_URL = "http://127.0.0.1:8000/api/order/" def submit(_): headers = {"Idempotency-Key": f"test-{_}"} payload = {"sku_id": 1, "qty": 1} resp = requests.post(ORDER_URL, json=payload, headers=headers, timeout=5) return resp.status_code with ThreadPoolExecutor(max_workers=50) as pool: statuses = list(pool.map(submit, range(100))) print("200 数量:", statuses.count(200)) print("库存剩余:", check_stock(1))

并发 50 跑 100 次后，如果库存剩余小于理论值，说明有更新丢失；如果200数量大于实际库存，说明接口没有真正校验。压测时还要盯 MySQL 的SHOW ENGINE INNODB STATUS \G输出，看到死锁计数上涨就检查select_for_update的顺序。最后再切到django.core.cache的 Redis 后段，确认缓存 key 过期时间没有超过支付回调重试的最长间隔。

本文还有配套的精品资源，点击获取