系統分頁設計完全指南:從基礎概念到進階實作

深入探討 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 分頁
  原因:自然的分割方式

實作檢查清單

  • 選擇合適的分頁策略
  • 建立必要的資料庫索引
  • 實作參數驗證和安全控制
  • 考慮快取策略
  • 提供清晰的分頁資訊
  • 處理邊界情況(空結果、最後一頁)
  • 優化深度分頁效能
  • 實作合適的錯誤處理
  • 考慮行動裝置體驗
  • 監控分頁效能指標

記住,分頁不只是技術問題,更是使用者體驗的關鍵。選擇合適的策略,才能在效能和體驗之間找到最佳平衡。


🔗 延伸閱讀

0%