如何编写数据分页查询脚本

wen 实用脚本 2026-07-01 204

本文目录导读：

如何编写数据分页查询脚本

分页查询核心概念
SQL 分页查询示例
Node.js + MySQL 实现
Python + SQLAlchemy 实现
Java Spring Boot 实现
前端分页组件示例 (React)
高级分页技巧
最佳实践建议

我来详细介绍如何编写数据分页查询脚本,包含不同数据库和后端语言的实现方式。

分页查询核心概念

分页查询通常需要两个参数：

page (当前页码)
pageSize (每页记录数)

计算关键值：

offset = (page - 1) × pageSize
limit = pageSize

SQL 分页查询示例

MySQL / PostgreSQL

-- 基础分页查询
SELECT * FROM users 
ORDER BY id 
LIMIT ? OFFSET ?;
-- 获取总记录数
SELECT COUNT(*) as total FROM users;
-- 完整分页查询
SELECT 
    u.*,
    (SELECT COUNT(*) FROM users) as total_count
FROM users u
ORDER BY u.id
LIMIT ? OFFSET ?;

SQL Server

-- SQL Server 2012+
SELECT * FROM users
ORDER BY id
OFFSET ? ROWS
FETCH NEXT ? ROWS ONLY;
-- 旧版本使用 ROW_NUMBER
WITH UserCTE AS (
    SELECT *,
        ROW_NUMBER() OVER (ORDER BY id) as RowNum
    FROM users
)
SELECT * FROM UserCTE
WHERE RowNum BETWEEN ? AND ?;

Oracle

SELECT * FROM (
    SELECT a.*, ROWNUM rn
    FROM (
        SELECT * FROM users ORDER BY id
    ) a
    WHERE ROWNUM <= ?
)
WHERE rn > ?;

Node.js + MySQL 实现

// pagination.js
const mysql = require('mysql2/promise');
class PaginationService {
    constructor(pool) {
        this.pool = pool;
    }
    async getPaginatedData(table, page = 1, pageSize = 10, conditions = {}) {
        const offset = (page - 1) * pageSize;
        // 构建查询条件
        let whereClause = '';
        let params = [];
        if (Object.keys(conditions).length > 0) {
            whereClause = 'WHERE ' + Object.keys(conditions)
                .map(key => `${key} = ?`)
                .join(' AND ');
            params = Object.values(conditions);
        }
        try {
            // 获取总记录数
            const [countResult] = await this.pool.execute(
                `SELECT COUNT(*) as total FROM ${table} ${whereClause}`,
                params
            );
            const total = countResult[0].total;
            // 获取分页数据
            const [rows] = await this.pool.execute(
                `SELECT * FROM ${table} ${whereClause} ORDER BY id LIMIT ? OFFSET ?`,
                [...params, pageSize, offset]
            );
            return {
                data: rows,
                pagination: {
                    page,
                    pageSize,
                    total,
                    totalPages: Math.ceil(total / pageSize),
                    hasNext: page < Math.ceil(total / pageSize),
                    hasPrev: page > 1
                }
            };
        } catch (error) {
            throw new Error(`Pagination failed: ${error.message}`);
        }
    }
}
// 使用示例
const pool = mysql.createPool({
    host: 'localhost',
    user: 'root',
    password: 'password',
    database: 'mydb'
});
const paginationService = new PaginationService(pool);
app.get('/api/users', async (req, res) => {
    try {
        const { page = 1, pageSize = 10 } = req.query;
        const result = await paginationService.getPaginatedData(
            'users', 
            parseInt(page), 
            parseInt(pageSize)
        );
        res.json(result);
    } catch (error) {
        res.status(500).json({ error: error.message });
    }
});

Python + SQLAlchemy 实现

# pagination.py
from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from dataclasses import dataclass
from typing import Optional, List, Any
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///database.db'
db = SQLAlchemy(app)
@dataclass
class PaginationResult:
    data: List[Any]
    pagination: dict
class PaginationMixin:
    @classmethod
    def paginate(cls, page: int = 1, page_size: int = 10, 
                 filters: Optional[dict] = None, order_by: str = 'id'):
        """通用分页方法"""
        query = cls.query
        # 应用过滤条件
        if filters:
            for key, value in filters.items():
                query = query.filter(getattr(cls, key) == value)
        # 计算总记录数
        total = query.count()
        # 获取分页数据
        offset = (page - 1) * page_size
        data = query.order_by(order_by).offset(offset).limit(page_size).all()
        return PaginationResult(
            data=data,
            pagination={
                'page': page,
                'page_size': page_size,
                'total': total,
                'total_pages': (total + page_size - 1) // page_size,
                'has_next': page < (total + page_size - 1) // page_size,
                'has_prev': page > 1
            }
        )
# 模型类
class User(db.Model, PaginationMixin):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100))
    email = db.Column(db.String(100))
class Product(db.Model, PaginationMixin):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100))
    price = db.Column(db.Float)
# API 路由
@app.route('/api/items/<model_name>')
def get_paginated_items(model_name):
    try:
        page = request.args.get('page', 1, type=int)
        page_size = request.args.get('page_size', 10, type=int)
        # 模型映射
        models = {
            'users': User,
            'products': Product
        }
        if model_name not in models:
            return jsonify({'error': 'Invalid model'}), 400
        model = models[model_name]
        # 构建过滤条件
        filters = {}
        if 'status' in request.args:
            filters['status'] = request.args['status']
        result = model.paginate(page, page_size, filters)
        # 序列化结果
        serialized_data = []
        for item in result.data:
            item_dict = {col.name: getattr(item, col.name) 
                        for col in item.__table__.columns}
            serialized_data.append(item_dict)
        return jsonify({
            'data': serialized_data,
            'pagination': result.pagination
        })
    except Exception as e:
        return jsonify({'error': str(e)}), 500

Java Spring Boot 实现

// PaginationService.java
@Service
public class PaginationService {
    @Autowired
    private JdbcTemplate jdbcTemplate;
    public PaginatedResult executePaginationQuery(
            String baseQuery, 
            String countQuery,
            Object[] params, 
            int page, 
            int pageSize,
            RowMapper<?> rowMapper) {
        int offset = (page - 1) * pageSize;
        // 获取总记录数
        int total = jdbcTemplate.queryForObject(countQuery, params, Integer.class);
        // 获取分页数据
        String paginatedQuery = baseQuery + " LIMIT ? OFFSET ?";
        Object[] paginatedParams = ArrayUtils.addAll(params, pageSize, offset);
        List<?> data = jdbcTemplate.query(paginatedQuery, paginatedParams, rowMapper);
        return new PaginatedResult(data, page, pageSize, total);
    }
}
// 更专业的 PageHelper 使用方式
@Service
public class UserService {
    public PageInfo<User> getUsers(int pageNum, int pageSize) {
        PageHelper.startPage(pageNum, pageSize);
        List<User> users = userMapper.selectAll();
        return new PageInfo<>(users);
    }
}

前端分页组件示例 (React)

// Pagination.jsx
import React, { useState, useEffect } from 'react';
const PaginatedList = () => {
    const [data, setData] = useState([]);
    const [pagination, setPagination] = useState({
        page: 1,
        pageSize: 10,
        total: 0,
        totalPages: 0
    });
    const [loading, setLoading] = useState(false);
    const fetchData = async (page = 1) => {
        setLoading(true);
        try {
            const response = await fetch(
                `/api/items/users?page=${page}&page_size=10`
            );
            const result = await response.json();
            setData(result.data);
            setPagination(result.pagination);
        } catch (error) {
            console.error('Fetch error:', error);
        } finally {
            setLoading(false);
        }
    };
    useEffect(() => {
        fetchData();
    }, []);
    const handlePageChange = (newPage) => {
        if (newPage >= 1 && newPage <= pagination.totalPages) {
            fetchData(newPage);
        }
    };
    return (
        <div>
            {loading ? (
                <div>Loading...</div>
            ) : (
                <>
                    <ul>
                        {data.map(item => (
                            <li key={item.id}>{item.name}</li>
                        ))}
                    </ul>
                    <PaginationControls
                        currentPage={pagination.page}
                        totalPages={pagination.totalPages}
                        onPageChange={handlePageChange}
                    />
                </>
            )}
        </div>
    );
};
// 分页控件组件
const PaginationControls = ({ currentPage, totalPages, onPageChange }) => {
    const pages = [];
    for (let i = 1; i <= totalPages; i++) {
        pages.push(i);
    }
    return (
        <div className="pagination">
            <button 
                onClick={() => onPageChange(currentPage - 1)}
                disabled={currentPage <= 1}
            >
                上一页
            </button>
            {pages.map(page => (
                <button
                    key={page}
                    onClick={() => onPageChange(page)}
                    className={page === currentPage ? 'active' : ''}
                >
                    {page}
                </button>
            ))}
            <button 
                onClick={() => onPageChange(currentPage + 1)}
                disabled={currentPage >= totalPages}
            >
                下一页
            </button>
        </div>
    );
};

高级分页技巧

游标分页（Cursor-based pagination）

-- 适用于大数据集，避免OFFSET性能问题
SELECT * FROM users 
WHERE id > ? 
ORDER BY id 
LIMIT ?;
-- 结合时间戳
SELECT * FROM posts 
WHERE created_at < ? 
ORDER BY created_at DESC 
LIMIT ?;

缓存优化

# 使用Redis缓存分页结果
import redis
import json
class CachedPagination:
    def __init__(self, redis_client):
        self.redis = redis_client
    def get_paginated_data(self, cache_key, page, page_size, query_func):
        cache_key = f"{cache_key}:page:{page}:size:{page_size}"
        # 尝试从缓存获取
        cached = self.redis.get(cache_key)
        if cached:
            return json.loads(cached)
        # 执行查询
        result = query_func(page, page_size)
        # 缓存结果（设置过期时间）
        self.redis.setex(cache_key, 300, json.dumps(result))
        return result