本文目录导读：
- 分页查询核心概念
- 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
最佳实践建议
- 索引优化：确保排序和WHERE条件字段有索引
- 限制最大页数：防止恶意请求
- 使用光标分页：处理大数据集
- 缓存热门页：提升性能
- 参数验证：防止SQL注入
- 统一响应格式：便于前端处理
这个框架可以根据你的具体需求进行调整和扩展,需要我详细解释某个特定部分吗？
