FastAPI + MySQL + Redis 短链接生成与访问统计系统（含二维码）
FastAPI MySQL Redis 短链接生成与访问统计系统含二维码涉及后端开发、缓存优化、数据统计和二维码生成等多个技术点。 核心功能模块功能说明短链接生成长URL → 唯一短码如abc123重定向跳转访问短链接 → 302跳转到原始URL访问统计记录每次点击的IP、UA、时间、来源二维码生成为每个短链接生成二维码图片过期管理可设置短链接有效期Redis加速热点短链接缓存降低数据库压力️ 数据库设计 (MySQL)-- 短链接主表 CREATE TABLE short_links ( id BIGINT AUTO_INCREMENT PRIMARY KEY, short_code VARCHAR(8) NOT NULL UNIQUE, -- 短码 original_url TEXT NOT NULL, -- 原始URL created_at DATETIME DEFAULT CURRENT_TIMESTAMP, expires_at DATETIME NULL, -- 过期时间NULL表示永久 is_active TINYINT DEFAULT 1, -- 是否启用 total_clicks INT DEFAULT 0, -- 总点击数冗余字段 INDEX idx_short_code (short_code), INDEX idx_expires (expires_at) ); -- 访问日志表用于统计分析 CREATE TABLE click_logs ( id BIGINT AUTO_INCREMENT PRIMARY KEY, short_code VARCHAR(8) NOT NULL, ip_address VARCHAR(45), user_agent TEXT, referer VARCHAR(500), country VARCHAR(100), device_type VARCHAR(20), -- PC/Mobile/Tablet clicked_at DATETIME DEFAULT CURRENT_TIMESTAMP, INDEX idx_code_time (short_code, clicked_at) );⚡ Redis缓存策略Key设计 - short:{code} → original_url (字符串TTL1小时) - stats:{code}:hourly → ZSET (每小时点击数member时间戳) - stats:{code}:daily → ZSET (每天点击数) 流程 1. 请求短链接 → 先查Redis 2. 命中 → 直接返回URL异步写日志到消息队列 3. 未命中 → 查MySQL写入Redis返回URL FastAPI 后端实现项目结构shortlink_service/ ├── main.py ├── models.py # SQLAlchemy模型 ├── schemas.py # Pydantic模型 ├── crud.py # 数据库操作 ├── utils.py # 工具函数短码生成、二维码 ├── routers/ │ ├── link.py # 短链接CRUD │ └── stats.py # 统计数据接口 └── templates/ └── qrcode.html # 二维码展示页面关键代码示例1. 短码生成算法 (utils.py)import string, random import hashlib import base62 def generate_short_code(url: str, length: int 6) - str: 方法一基于MD5截取 hash_obj hashlib.md5(url.encode()) hex_digest hash_obj.hexdigest() # 将16进制转为62进制0-9a-zA-Z decimal_val int(hex_digest[:8], 16) return base62.encode(decimal_val)[:length] def generate_random_code(length: int 6) - str: 方法二随机生成需检查冲突 chars string.ascii_letters string.digits return .join(random.choices(chars, klength))2. 创建短链接接口 (routers/link.py)from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from .. import crud, schemas, utils from ..database import get_db from ..redis_client import redis_client router APIRouter(prefix/api/v1) router.post(/shorten) async def create_short_link( req: schemas.ShortLinkCreate, db: Session Depends(get_db) ): # 1. 生成短码 code utils.generate_short_code(req.url) # 2. 检查冲突极小概率 existing crud.get_link_by_code(db, code) if existing: # 若冲突则重新生成加盐 code utils.generate_random_code() # 3. 写入数据库 link crud.create_short_link(db, code, req.url, req.expire_days) # 4. 预热缓存 redis_client.setex(fshort:{code}, 3600, req.url) return { short_code: code, short_url: fhttp://yourdomain.com/{code}, original_url: req.url, qrcode_url: f/api/v1/qrcode/{code} } router.get(/{short_code}) async def redirect_to_url(short_code: str, request: Request, db: Session Depends(get_db)): # 1. 从Redis获取 url redis_client.get(fshort:{short_code}) if not url: # 2. 从DB获取 link crud.get_active_link(db, short_code) if not link or (link.expires_at and link.expires_at datetime.utcnow()): raise HTTPException(status_code404, detailLink expired or not found) url link.original_url # 3. 写入缓存 redis_client.setex(fshort:{short_code}, 3600, url) # 4. 异步记录日志使用后台任务 background_tasks.add_task(log_click, short_code, request) # 5. 更新总点击数异步 background_tasks.add_task(crud.increment_clicks, db, short_code) # 6. 302重定向 return RedirectResponse(urlurl, status_code302)3. 访问统计接口 (routers/stats.py)router.get(/stats/{short_code}) async def get_stats(short_code: str, period: str 7d, db: Session Depends(get_db)): period: 24h / 7d / 30d # 获取总点击数 link crud.get_link_by_code(db, short_code) if not link: raise HTTPException(404) # 按时间粒度聚合 granularity hour if period 24h else day raw_data crud.get_click_timeline(db, short_code, period, granularity) # 设备分布 devices crud.get_device_distribution(db, short_code) # 地域分布需配合IP库 countries crud.get_country_distribution(db, short_code) return { total_clicks: link.total_clicks, timeline: raw_data, devices: devices, countries: countries }4. 二维码生成 (utils.py router)import qrcode from io import BytesIO from fastapi.responses import StreamingResponse def generate_qrcode(data: str) - BytesIO: qr qrcode.QRCode(box_size10, border4) qr.add_data(data) qr.make(fitTrue) img qr.make_image(fill_colorblack, back_colorwhite) buf BytesIO() img.save(buf, formatPNG) buf.seek(0) return buf router.get(/qrcode/{short_code}) async def get_qrcode(short_code: str): short_url fhttp://yourdomain.com/{short_code} buf generate_qrcode(short_url) return StreamingResponse(buf, media_typeimage/png) 数据看板ECharts示例你可以用FastAPI提供JSON数据前端用ECharts展示// 每小时点击量柱状图 fetch(/api/v1/stats/${code}?period24h) .then(res res.json()) .then(data { var chart echarts.init(document.getElementById(chart)); chart.setOption({ title: { text: 过去24小时点击趋势 }, xAxis: { type: category, data: data.timeline.map(t t.hour) }, yAxis: { type: value }, series: [{ type: bar, data: data.timeline.map(t t.count) }] }); }); 部署与优化建议方面建议高并发Nginx反向代理 Gunicorn/Uvicorn多worker防重复提交Redis分布式锁创建短链接时恶意攻击IP限流FastAPI middleware Redis计数器二维码美化使用qrcode.image.svg或myqr添加logo自定义短码允许用户自定义需校验唯一性批量生成支持CSV导入异步处理
