系統分頁設計完全指南:從基礎概念到進階實作
深入探討 Offset、Cursor、Seek 分頁的原理與最佳實踐
目錄
為什麼分頁如此重要?
想像一下,你正在瀏覽一個擁有百萬商品的電商網站,或是查看社交媒體上的無限動態。如果系統試圖一次載入所有資料,會發生什麼?
- 伺服器記憶體爆炸 💥
- 網路傳輸癱瘓 🌐
- 瀏覽器當機 💻
- 用戶體驗災難 😱
分頁(Pagination)是解決大數據集展示的核心技術,讓我們深入探討如何正確實作。
📊 分頁的核心挑戰
在深入技術細節前,先理解分頁要解決的問題:
挑戰 1:資料量
- 資料庫有 1000 萬筆記錄
- 每筆 1KB = 10GB 資料
- 不可能一次傳輸
挑戰 2:效能
- 查詢時間隨資料量增長
- 深度分頁效能退化
- 資源消耗控制
挑戰 3:一致性
- 資料即時變動
- 分頁過程中的新增/刪除
- 重複或遺漏問題
挑戰 4:用戶體驗
- 載入速度
- 導航便利性
- 位置記憶
🔧 三大分頁策略詳解
1. Offset-based Pagination(偏移分頁)
最直觀也最常見的分頁方式。
原理
-- 第 3 頁,每頁 20 筆
SELECT * FROM products
ORDER BY created_at DESC
LIMIT 20 OFFSET 40;
-- 計算公式
OFFSET = (page - 1) * page_size
實作範例
# Flask API 實作
from flask import Flask, request, jsonify
from math import ceil
@app.route('/api/products')
def get_products():
# 取得參數
page = int(request.args.get('page', 1))
per_page = int(request.args.get('per_page', 20))
# 防止不合理的值
page = max(1, page)
per_page = min(100, max(1, per_page)) # 限制最大 100
# 計算 offset
offset = (page - 1) * per_page
# 查詢資料
total_count = Product.query.count()
products = Product.query\
.order_by(Product.created_at.desc())\
.offset(offset)\
.limit(per_page)\
.all()
# 計算分頁資訊
total_pages = ceil(total_count / per_page)
has_prev = page > 1
has_next = page < total_pages
# 回傳結果
return jsonify({
'data': [p.to_dict() for p in products],
'pagination': {
'page': page,
'per_page': per_page,
'total': total_count,
'total_pages': total_pages,
'has_prev': has_prev,
'has_next': has_next
},
'links': {
'self': f'/api/products?page={page}&per_page={per_page}',
'first': f'/api/products?page=1&per_page={per_page}',
'last': f'/api/products?page={total_pages}&per_page={per_page}',
'prev': f'/api/products?page={page-1}&per_page={per_page}' if has_prev else None,
'next': f'/api/products?page={page+1}&per_page={per_page}' if has_next else None
}
})
優缺點分析
優點:
✅ 實作簡單直觀
✅ 可以跳頁(直接到第 N 頁)
✅ 適合靜態資料
✅ 用戶熟悉的介面
缺點:
❌ 深度分頁效能差(OFFSET 10000 很慢)
❌ 資料變動時會錯位
❌ 不適合即時更新的資料
效能問題詳解
-- 為什麼 OFFSET 大時很慢?
-- 查詢第 10000 頁
SELECT * FROM products
LIMIT 20 OFFSET 200000;
-- 資料庫必須:
-- 1. 掃描前 200000 筆記錄
-- 2. 跳過這些記錄
-- 3. 返回接下來的 20 筆
-- 時間複雜度:O(offset + limit)
2. Cursor-based Pagination(游標分頁)
使用指標追蹤位置,適合即時資料流。
原理
核心概念:
- 使用唯一標識作為游標
- 基於游標位置查詢下一批資料
- 避免 OFFSET 的效能問題
實作範例
import base64
import json
from datetime import datetime
class CursorPagination:
@staticmethod
def encode_cursor(timestamp, id):
"""編碼游標"""
cursor_data = {
'timestamp': timestamp.isoformat(),
'id': id
}
cursor_json = json.dumps(cursor_data)
cursor_bytes = cursor_json.encode('utf-8')
cursor_b64 = base64.b64encode(cursor_bytes).decode('utf-8')
return cursor_b64
@staticmethod
def decode_cursor(cursor):
"""解碼游標"""
try:
cursor_bytes = base64.b64decode(cursor.encode('utf-8'))
cursor_json = cursor_bytes.decode('utf-8')
cursor_data = json.loads(cursor_json)
return cursor_data
except:
return None
@app.route('/api/feed')
def get_feed():
limit = int(request.args.get('limit', 20))
cursor = request.args.get('cursor')
# 建立查詢
query = Post.query.order_by(
Post.created_at.desc(),
Post.id.desc() # 確保唯一排序
)
# 如果有游標,從游標位置開始
if cursor:
cursor_data = CursorPagination.decode_cursor(cursor)
if cursor_data:
cursor_time = datetime.fromisoformat(cursor_data['timestamp'])
cursor_id = cursor_data['id']
# 複合條件確保準確定位
query = query.filter(
db.or_(
Post.created_at < cursor_time,
db.and_(
Post.created_at == cursor_time,
Post.id < cursor_id
)
)
)
# 多查詢一筆來判斷是否有下一頁
posts = query.limit(limit + 1).all()
has_next = len(posts) > limit
posts = posts[:limit] # 只返回 limit 筆
# 生成下一頁游標
next_cursor = None
if has_next and posts:
last_post = posts[-1]
next_cursor = CursorPagination.encode_cursor(
last_post.created_at,
last_post.id
)
return jsonify({
'data': [p.to_dict() for p in posts],
'pagination': {
'limit': limit,
'has_next': has_next,
'next_cursor': next_cursor
}
})
優缺點分析
優點:
✅ 效能穩定(無深度分頁問題)
✅ 適合即時資料流
✅ 新資料不影響分頁
✅ 可以實現無限滾動
缺點:
❌ 無法跳頁
❌ 只能前進不能後退(或實作複雜)
❌ 游標可能失效
3. Seek/Keyset Pagination(鍵集分頁)
基於索引鍵的高效分頁方式。
原理
-- 傳統 OFFSET
SELECT * FROM products
WHERE category = 'electronics'
ORDER BY price ASC
LIMIT 20 OFFSET 1000; -- 慢!
-- Seek 方法
SELECT * FROM products
WHERE category = 'electronics'
AND price > 299.99 -- 上一頁最後一筆的價格
ORDER BY price ASC
LIMIT 20; -- 快!
實作範例
@app.route('/api/products/seek')
def get_products_seek():
limit = int(request.args.get('limit', 20))
# Seek 參數
last_price = request.args.get('last_price', type=float)
last_id = request.args.get('last_id', type=int)
# 建立基礎查詢
query = Product.query.order_by(
Product.price.asc(),
Product.id.asc() # 確保唯一排序
)
# 如果有 seek 參數,從該位置開始
if last_price is not None and last_id is not None:
# 複合條件處理相同價格的情況
query = query.filter(
db.or_(
Product.price > last_price,
db.and_(
Product.price == last_price,
Product.id > last_id
)
)
)
products = query.limit(limit).all()
# 準備下一頁的 seek 參數
next_params = None
if products:
last_product = products[-1]
next_params = {
'last_price': last_product.price,
'last_id': last_product.id
}
return jsonify({
'data': [p.to_dict() for p in products],
'pagination': {
'limit': limit,
'next_params': next_params
}
})
複雜排序的處理
class SeekPagination:
@staticmethod
def build_seek_filter(model, order_fields, last_values):
"""
建立複雜的 seek 過濾條件
order_fields: [(field, direction), ...]
last_values: 對應的值列表
"""
if not last_values:
return None
conditions = []
# 建立複合條件
for i in range(len(order_fields)):
field, direction = order_fields[i]
# 前面的欄位都相等
equal_conditions = []
for j in range(i):
equal_field, _ = order_fields[j]
equal_conditions.append(
getattr(model, equal_field) == last_values[j]
)
# 當前欄位的比較
if direction == 'asc':
current_condition = getattr(model, field) > last_values[i]
else:
current_condition = getattr(model, field) < last_values[i]
# 組合條件
if equal_conditions:
condition = db.and_(*equal_conditions, current_condition)
else:
condition = current_condition
conditions.append(condition)
return db.or_(*conditions)
# 使用範例
@app.route('/api/products/advanced-seek')
def advanced_seek():
# 複雜排序:類別降序 > 價格升序 > ID 升序
order_fields = [
('category', 'desc'),
('price', 'asc'),
('id', 'asc')
]
# 從請求取得上一頁最後的值
last_category = request.args.get('last_category')
last_price = request.args.get('last_price', type=float)
last_id = request.args.get('last_id', type=int)
last_values = None
if all(v is not None for v in [last_category, last_price, last_id]):
last_values = [last_category, last_price, last_id]
# 建立查詢
query = Product.query
# 套用排序
for field, direction in order_fields:
if direction == 'asc':
query = query.order_by(getattr(Product, field).asc())
else:
query = query.order_by(getattr(Product, field).desc())
# 套用 seek 過濾
if last_values:
seek_filter = SeekPagination.build_seek_filter(
Product, order_fields, last_values
)
query = query.filter(seek_filter)
# 執行查詢
products = query.limit(20).all()
return jsonify({
'data': [p.to_dict() for p in products]
})
🏗️ 進階分頁模式
1. Hybrid Pagination(混合分頁)
結合多種分頁方式的優點。
class HybridPagination:
"""
前幾頁用 offset(支援跳頁)
深度分頁自動切換到 cursor(效能優化)
"""
OFFSET_THRESHOLD = 1000 # 超過此值切換模式
@classmethod
def paginate(cls, query, page=None, cursor=None, per_page=20):
if cursor:
# Cursor 模式
return cls._cursor_paginate(query, cursor, per_page)
elif page and (page - 1) * per_page > cls.OFFSET_THRESHOLD:
# 深度分頁,建議使用 cursor
return {
'data': [],
'pagination': {
'error': 'Deep pagination not supported',
'suggestion': 'Use cursor-based pagination',
'max_offset': cls.OFFSET_THRESHOLD
}
}
else:
# Offset 模式
return cls._offset_paginate(query, page or 1, per_page)
2. Time-based Pagination(時間分頁)
特別適合時間序列資料。
@app.route('/api/logs')
def get_logs():
# 時間範圍參數
start_time = request.args.get('start_time')
end_time = request.args.get('end_time')
# 預設查詢最近一小時
if not start_time:
end_time = datetime.utcnow()
start_time = end_time - timedelta(hours=1)
else:
start_time = datetime.fromisoformat(start_time)
end_time = datetime.fromisoformat(end_time) if end_time else datetime.utcnow()
# 時間分片查詢
logs = LogEntry.query.filter(
LogEntry.timestamp.between(start_time, end_time)
).order_by(LogEntry.timestamp.desc()).limit(1000).all()
# 如果資料太多,建議縮小時間範圍
if len(logs) >= 1000:
suggested_end = logs[-1].timestamp
return jsonify({
'data': [log.to_dict() for log in logs],
'pagination': {
'has_more': True,
'suggested_next_query': {
'start_time': start_time.isoformat(),
'end_time': suggested_end.isoformat()
}
}
})
return jsonify({
'data': [log.to_dict() for log in logs],
'pagination': {
'has_more': False,
'time_range': {
'start': start_time.isoformat(),
'end': end_time.isoformat()
}
}
})
3. Infinite Scroll(無限滾動)
適合社交媒體和內容流。
// 前端實作
class InfiniteScroll {
constructor(options) {
this.container = options.container;
this.loadMore = options.loadMore;
this.threshold = options.threshold || 100;
this.loading = false;
this.hasMore = true;
this.cursor = null;
this.init();
}
init() {
// 監聽滾動事件
this.container.addEventListener('scroll', () => {
if (this.shouldLoadMore()) {
this.load();
}
});
// 載入第一頁
this.load();
}
shouldLoadMore() {
if (this.loading || !this.hasMore) return false;
const { scrollTop, scrollHeight, clientHeight } = this.container;
return scrollTop + clientHeight >= scrollHeight - this.threshold;
}
async load() {
this.loading = true;
try {
const response = await fetch(`/api/feed?cursor=${this.cursor || ''}`);
const data = await response.json();
// 渲染新內容
this.renderItems(data.data);
// 更新狀態
this.cursor = data.pagination.next_cursor;
this.hasMore = data.pagination.has_next;
} finally {
this.loading = false;
}
}
renderItems(items) {
items.forEach(item => {
const element = this.createItemElement(item);
this.container.appendChild(element);
});
}
}
🎯 不同場景的最佳選擇
場景分析對照表
場景 | 推薦方案 | 原因 |
---|---|---|
電商產品列表 | Offset | 用戶需要跳頁、看總數 |
社交動態 | Cursor | 即時更新、無限滾動 |
日誌查詢 | Time-based | 自然的時間分割 |
搜尋結果 | Offset + 限制 | 限制最大頁數 |
API 資料同步 | Cursor | 增量同步、斷點續傳 |
排行榜 | Offset | 需要知道具體排名 |
效能優化技巧
1. 資料庫索引優化
-- 為分頁查詢建立合適的索引
-- Offset 分頁
CREATE INDEX idx_created_at ON products(created_at DESC);
-- Seek 分頁(複合索引)
CREATE INDEX idx_category_price_id ON products(category, price, id);
-- 覆蓋索引(包含需要的欄位)
CREATE INDEX idx_products_listing ON products(
category,
price,
id,
name,
image_url
) WHERE status = 'active';
2. 快取策略
import redis
import hashlib
class PaginationCache:
def __init__(self):
self.redis = redis.Redis()
self.ttl = 300 # 5 分鐘
def get_cache_key(self, endpoint, params):
"""生成快取鍵"""
param_str = json.dumps(params, sort_keys=True)
param_hash = hashlib.md5(param_str.encode()).hexdigest()
return f"pagination:{endpoint}:{param_hash}"
def get_or_set(self, key, fetch_func):
"""快取模式"""
# 嘗試從快取取得
cached = self.redis.get(key)
if cached:
return json.loads(cached)
# 快取未命中,執行查詢
data = fetch_func()
# 寫入快取
self.redis.setex(
key,
self.ttl,
json.dumps(data)
)
return data
# 使用範例
@app.route('/api/products/cached')
def get_products_cached():
cache = PaginationCache()
# 建立快取鍵
params = {
'page': request.args.get('page', 1),
'category': request.args.get('category'),
'sort': request.args.get('sort', 'created_at')
}
cache_key = cache.get_cache_key('products', params)
def fetch_products():
# 實際的資料庫查詢
page = int(params['page'])
products = Product.query.paginate(
page=page,
per_page=20
)
return {
'data': [p.to_dict() for p in products.items],
'total': products.total
}
return jsonify(cache.get_or_set(cache_key, fetch_products))
3. 預載入與預取
class SmartPagination:
"""智慧分頁:預載入下一頁"""
@staticmethod
def paginate_with_prefetch(query, page, per_page):
# 同時查詢當前頁和下一頁
items = query.limit(per_page * 2).offset((page - 1) * per_page).all()
current_page = items[:per_page]
next_page_preview = items[per_page:per_page + 5] # 預覽下一頁前 5 筆
return {
'data': current_page,
'prefetch': {
'next_page_preview': next_page_preview,
'has_next': len(items) > per_page
}
}
🛡️ 分頁安全性考量
1. 參數驗證
from marshmallow import Schema, fields, validate
class PaginationSchema(Schema):
page = fields.Integer(
missing=1,
validate=validate.Range(min=1, max=10000) # 防止惡意深度分頁
)
per_page = fields.Integer(
missing=20,
validate=validate.Range(min=1, max=100) # 限制單頁大小
)
sort = fields.String(
missing='created_at',
validate=validate.OneOf(['created_at', 'price', 'name']) # 白名單
)
order = fields.String(
missing='desc',
validate=validate.OneOf(['asc', 'desc'])
)
# 使用
@app.route('/api/products/secure')
def get_products_secure():
schema = PaginationSchema()
try:
params = schema.load(request.args)
except ValidationError as err:
return jsonify({'errors': err.messages}), 400
# 安全的分頁查詢
return paginate_products(params)
2. 防止資源耗盡
class RateLimitedPagination:
"""限流的分頁"""
def __init__(self):
self.limiter = Limiter(
app,
key_func=lambda: get_remote_address(),
default_limits=["1000 per hour"]
)
@limiter.limit("10 per minute") # 深度分頁限流
def deep_pagination(self, page):
if page > 100:
# 深度分頁需要更嚴格的限流
raise TooManyRequests("Deep pagination is rate limited")
return self.paginate(page)
📱 前端整合最佳實踐
1. React 分頁元件
import React, { useState, useEffect } from 'react';
const PaginationComponent = ({ apiEndpoint }) => {
const [data, setData] = useState([]);
const [pagination, setPagination] = useState({});
const [loading, setLoading] = useState(false);
const fetchPage = async (page) => {
setLoading(true);
try {
const response = await fetch(
`${apiEndpoint}?page=${page}&per_page=20`
);
const result = await response.json();
setData(result.data);
setPagination(result.pagination);
} finally {
setLoading(false);
}
};
useEffect(() => {
fetchPage(1);
}, []);
const renderPaginationControls = () => {
const { page, total_pages, has_prev, has_next } = pagination;
return (
<div className="pagination-controls">
<button
onClick={() => fetchPage(1)}
disabled={page === 1}
>
首頁
</button>
<button
onClick={() => fetchPage(page - 1)}
disabled={!has_prev}
>
上一頁
</button>
<span>第 {page} / {total_pages} 頁</span>
<button
onClick={() => fetchPage(page + 1)}
disabled={!has_next}
>
下一頁
</button>
<button
onClick={() => fetchPage(total_pages)}
disabled={page === total_pages}
>
末頁
</button>
{/* 跳頁 */}
<input
type="number"
min="1"
max={total_pages}
placeholder="跳至"
onKeyPress={(e) => {
if (e.key === 'Enter') {
const targetPage = parseInt(e.target.value);
if (targetPage >= 1 && targetPage <= total_pages) {
fetchPage(targetPage);
}
}
}}
/>
</div>
);
};
return (
<div>
{loading ? (
<div>載入中...</div>
) : (
<>
<div className="data-list">
{data.map(item => (
<div key={item.id}>{item.name}</div>
))}
</div>
{renderPaginationControls()}
</>
)}
</div>
);
};
2. Vue 無限滾動
<template>
<div class="infinite-scroll-container" ref="container">
<div v-for="item in items" :key="item.id" class="item">
{{ item.name }}
</div>
<div v-if="loading" class="loading">
載入中...
</div>
<div v-if="!hasMore" class="end-message">
已經到底了
</div>
</div>
</template>
<script>
export default {
data() {
return {
items: [],
cursor: null,
loading: false,
hasMore: true
}
},
mounted() {
this.loadMore();
this.setupInfiniteScroll();
},
methods: {
async loadMore() {
if (this.loading || !this.hasMore) return;
this.loading = true;
try {
const params = new URLSearchParams();
if (this.cursor) {
params.append('cursor', this.cursor);
}
params.append('limit', 20);
const response = await fetch(`/api/feed?${params}`);
const data = await response.json();
this.items.push(...data.data);
this.cursor = data.pagination.next_cursor;
this.hasMore = data.pagination.has_next;
} finally {
this.loading = false;
}
},
setupInfiniteScroll() {
const options = {
root: this.$refs.container,
rootMargin: '100px',
threshold: 0.1
};
const observer = new IntersectionObserver((entries) => {
if (entries[0].isIntersecting) {
this.loadMore();
}
}, options);
// 觀察最後一個元素
this.$watch('items', () => {
this.$nextTick(() => {
const items = this.$refs.container.querySelectorAll('.item');
if (items.length > 0) {
observer.observe(items[items.length - 1]);
}
});
}, { immediate: true });
}
}
}
</script>
🎯 總結與建議
選擇指南
小數據集(< 10,000 筆):
推薦:Offset 分頁
原因:簡單直觀,效能可接受
中等數據集(10,000 - 1,000,000 筆):
推薦:Hybrid 分頁
原因:平衡效能與使用體驗
大數據集(> 1,000,000 筆):
推薦:Cursor/Seek 分頁
原因:效能優先
即時數據流:
推薦:Cursor 分頁
原因:處理新增資料
時間序列:
推薦:Time-based 分頁
原因:自然的分割方式
實作檢查清單
- 選擇合適的分頁策略
- 建立必要的資料庫索引
- 實作參數驗證和安全控制
- 考慮快取策略
- 提供清晰的分頁資訊
- 處理邊界情況(空結果、最後一頁)
- 優化深度分頁效能
- 實作合適的錯誤處理
- 考慮行動裝置體驗
- 監控分頁效能指標
記住,分頁不只是技術問題,更是使用者體驗的關鍵。選擇合適的策略,才能在效能和體驗之間找到最佳平衡。
🔗 延伸閱讀
- 📖 《High Performance MySQL》- 深入資料庫優化
- 📄 Slack’s Pagination Guide - API 分頁最佳實踐
- 🎥 Facebook’s Cursor Pagination - 大規模分頁實踐
- 💻 開源專案:Django Pagination - 成熟的分頁實作