php分页消息队列,vps主机和虚拟主机的区别,山东枣庄高防vps,vps免备案服务器宝塔面板搭建步骤,固定ip vps

🗣 Wd提问: php分页消息队列

🤖 Ai回答:
我来为您介绍PHP分页消息队列的实现方案。

1、基于Redis的消息队列分页

消息队列基础类

php
<?php
class RedisMessageQueue {
private $redis;
private $queueName;

public function __construct($queueName, $redisConfig = []) {
$this-queueName = $queueName;
$this-redis = new Redis();
$this-redis-connect($redisConfig['host'] ?? '127.0.0.1',
$redisConfig['port'] ?? 6379);
}

// 入队
public function push($message) {
$data = [
'id' = uniqid(),
'message' = $message,
'created_at' = time(),
'status' = 'pending'
];
return $this-redis-lpush($this-queueName, json_encode($data));
}

// 出队
public function pop() {
$data = $this-redis-rpop($this-queueName);
return $data ? json_decode($data, true) : null;
}

// 获取队列长度
public function getLength() {
return $this-redis-llen($this-queueName);
}
}

分页消息队列类

php
<?php
class PaginatedMessageQueue extends RedisMessageQueue {

// 获取分页消息
public function getMessages($page = 1, $limit = 10) {
$start = ($page 1) * $limit;
$end = $start + $limit 1;

// 获取指定范围的消息
$messages = $this-redis-lrange($this-queueName, $start, $end);
$total = $this-getLength();

$result = [];
foreach ($messages as $message) {
$result[] = json_decode($message, true);
}

return [
'data' = array_reverse($result), // 反转以显示最新消息在前
'total' = $total,
'page' = $page,
'limit' = $limit,
'pages' = ceil($total / $limit)
];
}

// 根据状态筛选消息
public function getMessagesByStatus($status, $page = 1, $limit = 10) {
$allMessages = $this-redis-lrange($this-queueName, 0, -1);
$filteredMessages = [];

foreach ($allMessages as $message) {
$data = json_decode($message, true);
if ($data['status'] === $status) {
$filteredMessages[] = $data;
}
}

$total = count($filteredMessages);
$start = ($page 1) * $limit;
$pagedMessages = array_slice(array_reverse($filteredMessages), $start, $limit);

return [
'data' = $pagedMessages,
'total' = $total,
'page' = $page,
'limit' = $limit,
'pages' = ceil($total / $limit)
];
}

// 更新消息状态
public function updateMessageStatus($messageId, $status) {
$allMessages = $this-redis-lrange($this-queueName, 0, -1);
$this-redis-del($this-queueName);

foreach ($allMessages as $message) {
$data = json_decode($message, true);
if ($data['id'] === $messageId) {
$data['status'] = $status;
$data['updated_at'] = time();
}
$this-redis-rpush($this-queueName, json_encode($data));
}
}
}

2、基于数据库的消息队列分页

数据库表结构

sql
CREATE TABLE message_queue (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
message_id VARCHAR(32) NOT NULL UNIQUE,
message TEXT NOT NULL,
status ENUM('pending', 'processing', 'completed', 'failed') DEFAULT 'pending',
priority TINYINT DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
INDEX idx_status (status),
INDEX idx_priority_created (priority, created_at)
);

数据库消息队列类

php
<?php
class DatabaseMessageQueue {
private $pdo;
private $tableName;

public function __construct($pdo, $tableName = 'message_queue') {
$this-pdo = $pdo;
$this-tableName = $tableName;
}

// 入队
public function push($message, $priority = 0) {
$sql = INSERT INTO {$this-tableName} (message_id, message, priority) VALUES (?, ?, ?);
$stmt = $this-pdo-prepare($sql);
return $stmt-execute([uniqid(), json_encode($message), $priority]);
}

// 出队
public function pop() {
$this-pdo-beginTransaction();
try {
// 获取优先级最高的待处理消息
$sql = SELECT * FROM {$this-tableName} WHERE status = 'pending' ORDER BY priority DESC, created_at ASC LIMIT 1 FOR UPDATE;
$stmt = $this-pdo-query($sql);
$message = $stmt-fetch(PDO::FETCH_ASSOC);

if ($message) {
// 更新状态为处理中
$updateSql = UPDATE {$this-tableName} SET status = 'processing', updated_at = NOW() WHERE id = ?;
$updateStmt = $this-pdo-prepare($updateSql);
$updateStmt-execute([$message['id']]);

$this-pdo-commit();
return $message;
}

$this-pdo-commit();
return null;
} catch (Exception $e) {
$this-pdo-rollback();
throw $e;
}
}

// 分页获取消息
public function getMessages($page = 1, $limit = 10, $filters = []) {
$offset = ($page 1) * $limit;

$where = 1=1;
$params = [];

// 状态筛选
if (!empty($filters['status'])) {
$where .= AND status = ?;
$params[] = $filters['status'];
}

// 优先级筛选
if (isset($filters['priority'])) {
$where .= AND priority = ?;
$params[] = $filters['priority'];
}

// 时间范围筛选
if (!empty($filters['start_time'])) {
$where .= AND created_at = ?;
$params[] = $filters['start_time'];
}

if (!empty($filters['end_time'])) {
$where .= AND created_at <= ?;
$params[] = $filters['end_time'];
}

// 获取总数
$countSql = SELECT COUNT(*) FROM {$this-tableName} WHERE {$where};
$countStmt = $this-pdo-prepare($countSql);
$countStmt-execute($params);
$total = $countStmt-fetchColumn();

// 获取分页数据
$order = $filters['order'] ?? 'created_at DESC';
$sql = SELECT * FROM {$this-tableName} WHERE {$where} ORDER BY {$order} LIMIT ? OFFSET ?;
$params[] = $limit;
$params[] = $offset;

$stmt = $this-pdo-prepare($sql);
$stmt-execute($params);
$messages = $stmt-fetchAll(PDO::FETCH_ASSOC);

return [
'data' = $messages,
'total' = (int)$total,
'page' = (int)$page,
'limit' = (int)$limit,
'pages' = ceil($total / $limit)
];
}

// 更新消息状态
public function updateStatus($messageId, $status) {
$sql = UPDATE {$this-tableName} SET status = ?, updated_at = NOW() WHERE message_id = ?;
$stmt = $this-pdo-prepare($sql);
return $stmt-execute([$status, $messageId]);
}

// 批量更新状态
public function batchUpdateStatus($messageIds, $status) {
$placeholders = str_repeat('?,', count($messageIds) 1) 、'?';
$sql = UPDATE {$this-tableName} SET status = ?, updated_at = NOW() WHERE message_id IN ({$placeholders});
$params = array_merge([$status], $messageIds);

$stmt = $this-pdo-prepare($sql);
return $stmt-execute($params);
}
}

3、使用示例

Redis分页队列使用

php
<?php
// 初始化队列
$queue = new PaginatedMessageQueue('user_notifications', [
'host' = '127.0.0.1',
'port' = 6379
]);

// 添加消息
$queue-push(['user_id' = 1, 'content' =

Ai作答

📣 商家广告：