企业官网建设流程全解析

简介：基于Python+Flask+Redis的餐厅菜品推荐系统是一份完整的高分毕业设计项目，面向软件工程、计算机科学、人工智能等专业的在校学生与教师，也适合作为课程设计、毕业设计、作业或项目立项演示。系统包含数据爬取、后台服务、前端展示与推荐可视化等模块，代码已通过运行测试，可直接部署或二次开发。压缩包共99个文件、约5.76MB，核心类型有Python脚本、HTML页面、JavaScript与CSS样式，另含城市与商家数据表格、界面截图及基于ECharts的可视化配置，目录结构清晰，便于按模块深入学习。资源附带详细文档和全部资料，爬虫脚本覆盖大众点评、美团等平台，前端提供菜品排行、关键词云、区域搜索等视图，能够完整呈现推荐系统的工作流程。已有104人学习下载，适合需要快速搭建推荐系统毕设或理解Flask与Redis整合开发的人群参考。

1. 一份 Python + Flask + Redis 能扛住的餐厅推荐系统，其实先把边界划清

餐厅菜品推荐和电商推荐最大的区别在于：用户没有“购物车”，决策链路短，而且绝大多数人走进一家店之前并不清楚自己今天想吃什么。这个场景下，推荐系统的任务不是把用户拉回平台，而是在用户打开点餐页的几秒内给出足够合理的候选集，否则用户直接滑走。同时它的流量峰值高度集中——午市和晚市前后各一小时，平时的查询量低得可怜，这就决定了我们在做架构选型时不应该追求复杂的分布式计算，而要把一个单机应用打磨到极致。

用 Python + Flask + Redis 做这个系统，本质上是拿 Flask 只做 HTTP 层的工作：参数解析、鉴权、结果序列化，真正的推荐计算逻辑放在独立的 service 层里，Redis 承担两层职责——缓存用户近期的行为序列，以及存储通过离线计算产出的候选集和商品画像。这套结构的好处是，即便你在毕业设计或者小型项目中只部署一台服务器，用户请求的响应时间也能稳定压在 50ms 以内，因为你的推荐计算根本不涉及数据库扫描。

这套方案的适用人群很明确：对 Flask 路由和蓝图已经有基本概念的开发者，想了解 Redis 不只有缓存这一种用法的人，以及需要写一份“可运行、可说清、可答辩”的完整项目的人。接下来的内容会把数据模型、存储结构、接口实现和踩坑点一条线拉通，你可以照着逐步搭出来。

2. 菜品推荐的数据结构选型：Redis 里放什么、不放什么

2.1 用户、菜品、评分三个基础模型怎么落到 Redis 里

先说结论：Redis 里只放两类数据——用户近 N 天的行为序列和离线计算好的候选菜品集合。用户表、菜品表、订单表仍然放在 MySQL 中，因为它们是关系型数据，需要事务和复杂的条件查询。如果你想让 MySQL 里的数据和 Redis 里的结构对齐，常见的做法是在菜品主表上冗余一个category字段和一个avg_rating字段，这样离线任务在拉取数据时可以少做一次 JOIN。

用户行为序列，我推荐用两种结构组合使用：

• user:recent:{user_id}，类型为 list，每次用户点餐后从左侧推入一个菜品 ID，表示最近操作的事件流。
• user:pref:{user_id}，类型为 hash，字段是品类 ID，值是这个用户在过去 7 天对某一品类的点击次数。

为什么不直接用 MySQL 里的订单表来实时计算用户的偏好？因为每次推荐请求都要实时聚合订单表，高峰时段会拖垮数据库。Redis 里的这两种结构是异步更新的——用户在 Flask 接口里下单成功后，接口只做一件事：向 Redis 写入操作日志，然后返回成功，后续由后台脚本统一刷入这两个结构。这个过程叫「先写缓存，后进数据库」，在推荐场景下是可接受的短暂不一致。

2.2 ZSet 是菜品热榜的底层实现，但小心分数衰减

菜品热度榜是餐厅推荐系统最常用也最好解释的推荐依据：把一个区域或一个店内近 24 小时被下单次数最多的菜品捞出来。用 Redis 的 ZSet 做这件事非常自然，member 是菜品 ID，score 是一个组合分值，比如：

score = 点赞数 * 3 + 点单数 * 2 + 浏览量 * 1

如果你永远只累加不衰减，会出现一个问题：老菜品的分数只增不减，新菜永远上不了榜。常见的衰减策略是：每天凌晨 3 点，用一个定时任务读取当前 ZSet 里所有成员，把每个分数乘以 0.6，低于阈值的成员直接移除。这个衰减系数你可以根据菜品味型的新鲜周期去调整。

# 查看前 50 名菜品，带分数，供后台运营确认榜单是否合理 ZREVRANGE dish:hot:today 0 49 WITHSCORES

再往前一步，如果你希望热榜是按店铺维度输出的，就把 ZSet 的 key 设计成dish:hot:{shop_id}:{date}，这样每个店铺的榜单之间互不干扰，后续分页和淘汰也方便。

2.3 候选菜品集合每天刷一次，线上请求只读缓存

离线计算候选集这一步，用 Python 脚本从 MySQL 里把菜品数据拉出来，对每个用户生成一个推荐列表，写入 Redis，key 形如rec:cand:{user_id}，类型是 string，内容为 JSON 数组，里面放 100 个菜品 ID 和对应的分数。为什么是 100 而不是 20？因为线上接口拿到这 100 个后还要做过滤、去重和基于实时上下文的排序，候选集多一些，兜底能力就强。

import redis import json r = redis.Redis(host='127.0.0.1', port=6379, db=0, decode_responses=True) candidates = [] # 此处省略从 MySQL 拉取数据并计算相似度的过程 # candidates 的结构是 [{"dish_id": 101, "score": 4.5}, ...] r.set('rec:cand:10001', json.dumps(candidates), ex=86400)

这段代码里有两个参数值得说明：ex=86400代表这个 key 存活 24 小时，即每天候选集只重建一次；decode_responses=True会在读取时直接返回字符串而不是字节，省去每次手动 decode 的麻烦。如果你们的项目里 Redis 数据量较大，可以给这个脚本加上try...except捕获连接超时，避免一次 OFFLINE 任务失败把整个系统的缓存全清掉。这部分数据仍然可以加上一个可用的过期时间，防止内存不断膨胀。

3. Flask 薄接口层：路由、校验、参数协议与推荐引擎的粘合方式

3.1 接口就三个：首页推荐、按菜系筛选、用户反馈上报

一个推荐系统做得再花哨，对外暴露的接口最好控制在三个以内。原因很简单：接口数量越少，鉴权和限流的成本越低，参数校验和异常分支需要覆盖的情况越少。我会给这套系统定义三个核心路由：

from flask import Flask, request, jsonify from service.recommend import RecommendationEngine app = Flask(__name__) engine = RecommendationEngine() @app.route('/api/recommend', methods=['GET']) def recommend(): user_id = request.args.get('user_id', type=int) size = request.args.get('size', default=20, type=int) shop_id = request.args.get('shop_id', default=0, type=int) if not user_id: return jsonify({'code': 400, 'msg': 'missing user_id'}) items = engine.recommend(user_id, shop_id, size) return jsonify({'code': 0, 'data': items})

这里有一个接口设计上的关键决定：user_id是必填参数，shop_id是可选的，默认 0 表示全平台推荐。为什么不支持匿名用户？因为匿名推荐只能退化为热榜，而热榜本身就包含在推荐结果里，所以直接要求前端在用户进入点餐页时完成静默登录，让每个请求都带身份信息，后端才有机会“千人千面”。size参数必须做上限控制，稍后我会说明为什么。

3.2 service 层的推荐编排，别让接口函数写成一坨

很多毕业设计失败在把推荐逻辑直接写在路由函数里——几十个 if else 判断用户有没有历史行为、Redis 有没有缓存、要不要走冷启动。正确的做法是把路由和推荐引擎拆开，引擎内部再分成三步：读缓存、算分、再过滤。

class RecommendationEngine: def __init__(self): self.redis = redis.Redis(host='127.0.0.1', port=6379, db=0, decode_responses=True) def recommend(self, user_id, shop_id, size=20): cache_key = f"rec:cand:{user_id}" candidates = self.redis.get(cache_key) if candidates is None: candidates = self._cold_start() else: candidates = json.loads(candidates) # 过滤本店下架的菜品，可以做成独立过滤链 candidates = self._filter_sold_out(candidates, shop_id) # 截断后返回，保证前端拿到的数量与参数一致 return candidates[:size]

_filter_sold_out是我预留的过滤钩子，它内部会读取当日的菜品上下架列表，判断候选集里哪些菜品该被剔除。这样做的意义是，候选集是凌晨算好的，但上午十点可能有一批菜被下架了，如果你不在这层过滤，用户就会看到无法下单的菜。这个钩子的数据来源可以用 Redis 的 set 保存下架菜品 ID，查询时走SISMEMBER，复杂度是 O(1)。

3.3 请求参数里的一个陷阱：size 必须限长

不少初学者会让size参数直接透传到推荐引擎内部，然后整个列表切片完事。一旦有人恶意传size=1000000，引擎会尝试把一个超长列表读进内存，Redis 内存和 Flask 进程同时遭殃。正确做法是加一个显式的边界控制：

size = min(max(size, 5), 50)

这里min和max组合的含义是：推荐结果最少返回 5 条，最多返回 50 条。用户传了 1，你也有 5 条兜底；用户传了 99999，你只给 50 条。这比if size > 50: size = 50更简洁，而且不会让后续维护的人漏掉下限边界。你可以在 API 文档里注明 50 条是平台规则决定的——一屏最多展示 20 条，下拉刷新两次就应该触底，多出来的数据只会增加加载时间。

3.4 Flask 集成 Redis 时的连接管理问题

Flask 开发模式下，每一个请求都会创建新的线程，如果你的代码里每次推荐请求都执行redis.Redis(host='...')，会建立大量无用的连接，Redis 端默认连接数吃紧以后会直接拒绝服务。正确的做法是全局只维护一个连接池，或者在应用启动时初始化一次。

from redis import ConnectionPool pool = ConnectionPool(host='127.0.0.1', port=6379, db=0, max_connections=20) def get_redis(): return redis.Redis(connection_pool=pool)

max_connections=20是并发上限，配合 SQLite 或 MySQL 连接池一起用，能保证高并发下 Flask 的线程和 Redis 连接数保持匹配。如果你的项目部署环境是 Docker 里两个容器，Redis 容器和 Flask 容器使用自定义网络互相访问，这里的 host 参数要改成服务名而不是127.0.0.1，这是本地环境跑通之后第一个容易踩的部署坑。用 Redis Desktop Manager 连上去看一眼也能帮你定位是网络不通还是密码不对。

4. 缓存一致性、冷启动与降级兜底：推荐系统真正吃时间的部分

4.1 用户反馈上报接口如何保证 Redis 和 MySQL 不出现走样的数据

推荐系统需要用户反馈来迭代：用户点了哪个菜、在哪个菜上停留了多久、最终下单了哪个。Flask 里这个接口叫 feedback，接收三个参数：user_id、dish_id、action。action 的取值有三种：view、click、order。这个接口要做的不是立刻修改推荐结果，而是把事件写入 Redis 的 list 结构，作为异步任务的输入源。

@app.route('/api/feedback', methods=['POST']) def feedback(): data = request.get_json() user_id = data.get('user_id') dish_id = data.get('dish_id') action = data.get('action') if action not in ('view', 'click', 'order'): return jsonify({'code': 400, 'msg': 'invalid action'}) event_key = 'event:feedback' r.lpush(event_key, {'user_id': user_id, 'dish_id': dish_id, 'action': action, 'ts': int(time.time())}) # 同步更新用户偏好 hash category = get_dish_category(dish_id) r.hincrby(f'user:pref:{user_id}', category, 1) return jsonify({'code': 0})

这个接口的精髓在hincrby而不是hset——前者是原子自增，多个请求同时上报时不会互相覆盖；后者需要先读后写，在并发场景下会丢数据。另外这里的event:feedback列表生产端写入的已经是 JSON 字符串，后台脚本消费时直接读取即可，不要在这里做二次序列化，减少出错面。

4.2 冷启动用户怎么推荐：基于品类偏好的缺省策略

新用户没有任何行为数据时，推荐系统不能直接返回热榜，因为热榜里的菜可能全是川菜，而用户完全不吃辣。冷启动的常见做法是：先给用户展示品类分布均衡的候选集，等用户产生三到五次浏览行为以后，再切到个性化推荐。实现时可以在user:pref不存在的情况下，用一个「通用候选」key 来兜底。

def _cold_start(self, shop_id=0, size=20): key = f"rec:cold:{shop_id}" candidates = self.redis.get(key) if candidates is None: candidates = self._build_cold_start(shop_id) self.redis.set(key, json.dumps(candidates), ex=3600) return candidates[:size]

_build_cold_start里的逻辑是从 MySQL 里按菜系分组，每个菜系取点赞最高的前 5 个菜，合并后打乱顺序，这样结果里至少有 40% 的菜不属于热门品类。为什么要打乱？因为排序太固定会让运营觉得“推荐系统没有脑子”，而且打乱后不同用户看到的第一屏不同，做 A/B 测试时更有区分度。这个通用候选的过期时间设为 3600 秒，也就是每小时自动重算一次，菜品上新后最迟一小时内就能进入新用户的首屏。

4.3 Redis 内存爆炸的 Guard：给候选集加 TTL，给用户序列加长度上限

餐饮点餐系统虽然数据量不如电商大，但是 Redis 里存了user:recent、user:pref、rec:cand、dish:hot之后，如果长期不清理，内存会一路飙升。典型的失控场景是：user:recent这个 list 只增不减，老用户点了一百次餐，list 里躺着一百个菜品 ID，而真正有用的只有最近 20 个。

解决办法是用LTRIM在每次写入后立刻裁剪列表长度：

def record_recent(user_id, dish_id): key = f"user:recent:{user_id}" r.lpush(key, dish_id) r.ltrim(key, 0, 19)

LTRIM key 0 19表示只保留列表的第 0 到第 19 个元素，其他全部删除。每次点餐最多产生两条记录，裁剪后这个 key 的占用空间就是恒定的，不会随使用时间膨胀。同样的道理也适用于event:feedback这个列表，可以启动一个定时任务，把 7 天前的原始事件删掉，避免占用过多内存。

4.4 Redis 挂掉之后的降级链路

虽然 Redis 正常情况下很稳定，但部署环境总有意外：内存满了触发淘汰策略、进程被 OOM Killer 杀掉、网络分区导致连接超时。推荐系统如果在 Redis 挂掉的时候直接返回 500，用户体验会非常差。常见的做法是加一个降级开关：从 Redis 读不到数据时，直接读取本地文件里的快照，或者返回一个由程序算出的简单热榜。

class RecommendationEngine: def recommend_with_fallback(self, user_id, size=20): try: return self.recommend(user_id, size) except redis.ConnectionError: return self._local_snapshot(size)

本地快照的实现不需要额外引入 JSON 文件管理系统，直接在同目录放一个snapshot.json，离线任务算完候选集后同时写 Redis 和本地文件。降级时返回的推荐结果可能不够新鲜，但至少在系统故障期间用户还能看到菜品，不至于白屏。

5. 用 Redis 原生命令实现榜单分页与按菜系换一换

5.1 转存 ZSet 实现翻页，而不是用 ZRANGE 全量拉取

热榜或推荐结果超过一屏之后，前端会做滚动加载。如果每次都调用ZREVRANGE key 0 -1，会把整张榜单全部传输给前端，数据量大时既浪费带宽又拖慢响应。更合理的做法是使用ZREVRANGE的分页参数直接从 Redis 取某一页，或者在后续需要做复杂过滤时转存到本地再操作。对于纯展示场景，按页取完全没有问题：

ZREVRANGE dish:hot:today 20 39 WITHSCORES

这条命令的含义是取热度排行第 21 名到第 40 名之间的数据，带分数返回。对用户来说是第二页的内容。但有一个坑：第一页请求和第二页请求之间如果有新的点单发生，这个榜单的排名会变，用户可能看到同一道菜出现在两页里。要避免这个体验问题，最简单的方式是前端在进入推荐页时一次性向后端要 50 条，然后本地做滚动分页，而不是每次滚动都发起新请求。

5.2 用 Lua 脚本原子化完成推荐结果的个性化排序

推荐引擎把候选集算出来之后，还需要根据用户的实时偏好微调排序。比如用户最近三天点过三次川菜，那么候选集里川菜的权重应该适当上浮。这个上浮如果分两步做——先从 Redis 读偏好、再在 Python 里加权——会出现并发下数据不一致的问题。用 Redis 的 Lua 脚本可以一次完成读取和计算，整个过程不会被打断。

local pref = redis.call('HGETALL', KEYS[1]) local cand = redis.call('GET', KEYS[2]) -- 此处省略对 cand 解析并加权的逻辑 return cand

在 Python 里调用这段脚本时，需要把两个 key 作为参数传进去，Redis 在执行脚本期间会阻塞其他命令，所以脚本只适合做微秒级的快速操作。推荐结果的整体个性化排序还是放回 Python 层做，Lua 只处理「必须原子」的那一部分，否则复杂计算会阻塞 Redis 主线程，反而拖累其他接口。

5.3 按菜系“换一批”的实践：用临时 key 隔离不同页的随机序列

用户经常点「换一批」，本质上是希望看到同品类下不同的菜品。实现上不要用SRANDMEMBER直接随机——因为每次刷新随机结果可能重复太多，用户会觉得没变化。正确的做法是给每个用户维护一个「已展示队列」，每次换一批时从候选池里剔除已展示过的 ID。

def refresh_by_category(user_id, category_id, size=10): pool_key = f"cand:cat:{category_id}" shown_key = f"shown:user:{user_id}:cat:{category_id}" pool = set(r.smembers(pool_key)) shown = r.smembers(shown_key) available = list(pool - shown) random.shuffle(available) picked = available[:size] if picked: r.sadd(shown_key, *picked) r.expire(shown_key, 1800) return picked

这段代码里r.sadd会把本次返回的菜品 ID 追加到已展示集合里，下一次换一批时这些菜就被过滤掉了。expire设了 1800 秒，意思是半小时之后这个集合自动清空，用户可以再次看到初始的那些菜——因为在真实场景里，用户半小时后多半已经忘了之前看过什么，没有必要保留永久性的去重记录。如果你想做得更精细，就把这些已展示的 ID 按日期切分，比如shown:user:10001:20250321，方便做每日维度的手工排查与数据复盘。

5.4 回归测试：如何验证推荐结果是不是真的在变好

系统上线前需要验证一件事：修改相似度算法或权重参数后，推荐结果评分是否真的优于旧版本。最直接的办法是记录每个推荐位的曝光量和点击量，计算 CTR（点击率），公式为点击次数 / 曝光次数。把旧版本日志和新版本日志各存一份，对比同一个用户在不同版本下的点击率变化。

FLask 接口里可以在返回推荐结果时带上version参数，前端在曝光时把 version 一起上报。后端的统计脚本从event:feedback列表里读取数据，按 version 分组，每日生成一张 CTR 报表。当新版 CTR 稳定高于旧版两个百分点以上，就说明这次改动是正向的；如果数据持平或下降，就需要回滚权重参数，而不是继续叠加新规则。这套验证流程能让你的系统迭代从「拍脑袋调参」变成「数据说了算」，而这也正是区分一份高分毕业设计与一个能上线运行的真实系统之间最重要的分界点。

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