本文目录导读：
我来为您设计一个完整的 PHP 未读消息系统。
数据库设计
消息表结构
-- 用户表（已有）
CREATE TABLE `users` (
`id` INT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
`username` VARCHAR(50) NOT NULL,
`created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP
) ENGINE=InnoDB;
-- 消息表
CREATE TABLE `messages` (
`id` BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
`from_user_id` INT UNSIGNED NOT NULL,
`to_user_id` INT UNSIGNED NOT NULL,
`content` TEXT NOT NULL,
`message_type` TINYINT DEFAULT 1 COMMENT '1:私信 2:系统通知 3:评论回复',
`created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
INDEX `idx_to_user` (`to_user_id`, `created_at`),
INDEX `idx_from_user` (`from_user_id`),
FOREIGN KEY (`from_user_id`) REFERENCES `users`(`id`),
FOREIGN KEY (`to_user_id`) REFERENCES `users`(`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
-- 消息状态表（优化查询性能）
CREATE TABLE `message_status` (
`user_id` INT UNSIGNED NOT NULL,
`message_id` BIGINT UNSIGNED NOT NULL,
`is_read` TINYINT(1) DEFAULT 0,
`read_at` TIMESTAMP NULL,
PRIMARY KEY (`user_id`, `message_id`),
INDEX `idx_unread` (`user_id`, `is_read`),
FOREIGN KEY (`user_id`) REFERENCES `users`(`id`),
FOREIGN KEY (`message_id`) REFERENCES `messages`(`id`)
) ENGINE=InnoDB;
-- 用户消息计数表（优化未读计数）
CREATE TABLE `user_message_counters` (
`user_id` INT UNSIGNED PRIMARY KEY,
`unread_count` INT UNSIGNED DEFAULT 0,
`last_message_id` BIGINT UNSIGNED DEFAULT 0,
`updated_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
FOREIGN KEY (`user_id`) REFERENCES `users`(`id`)
) ENGINE=InnoDB;
核心类设计
MessageService.php
<?php
class MessageService {
private $pdo;
private $cache;
public function __construct(PDO $pdo, $cache = null) {
$this->pdo = $pdo;
$this->cache = $cache;
}
/**
* 发送消息
*/
public function sendMessage($fromUserId, $toUserId, $content, $messageType = 1) {
try {
$this->pdo->beginTransaction();
// 插入消息记录
$sql = "INSERT INTO messages (from_user_id, to_user_id, content, message_type)
VALUES (?, ?, ?, ?)";
$stmt = $this->pdo->prepare($sql);
$stmt->execute([$fromUserId, $toUserId, $content, $messageType]);
$messageId = $this->pdo->lastInsertId();
// 更新消息状态
$sql = "INSERT INTO message_status (user_id, message_id) VALUES (?, ?)";
$stmt = $this->pdo->prepare($sql);
$stmt->execute([$toUserId, $messageId]);
// 更新未读计数
$this->incrementUnreadCount($toUserId);
$this->pdo->commit();
// 推送实时通知（可选）
$this->pushNotification($toUserId);
return $messageId;
} catch (Exception $e) {
$this->pdo->rollBack();
throw $e;
}
}
/**
* 获取用户未读消息列表
*/
public function getUnreadMessages($userId, $limit = 20, $offset = 0) {
$sql = "SELECT m.*, u.username
FROM messages m
INNER JOIN message_status ms ON m.id = ms.message_id
INNER JOIN users u ON m.from_user_id = u.id
WHERE ms.user_id = ? AND ms.is_read = 0
ORDER BY m.created_at DESC
LIMIT ? OFFSET ?";
$stmt = $this->pdo->prepare($sql);
$stmt->execute([$userId, $limit, $offset]);
return $stmt->fetchAll(PDO::FETCH_ASSOC);
}
/**
* 标记单条消息为已读
*/
public function markAsRead($userId, $messageId) {
$sql = "UPDATE message_status
SET is_read = 1, read_at = NOW()
WHERE user_id = ? AND message_id = ? AND is_read = 0";
$stmt = $this->pdo->prepare($sql);
$stmt->execute([$userId, $messageId]);
if ($stmt->rowCount() > 0) {
$this->decrementUnreadCount($userId);
return true;
}
return false;
}
/**
* 标记所有消息为已读
*/
public function markAllAsRead($userId) {
$sql = "UPDATE message_status
SET is_read = 1, read_at = NOW()
WHERE user_id = ? AND is_read = 0";
$stmt = $this->pdo->prepare($sql);
$stmt->execute([$userId]);
// 重置计数器
$sql = "UPDATE user_message_counters SET unread_count = 0 WHERE user_id = ?";
$stmt = $this->pdo->prepare($sql);
$stmt->execute([$userId]);
return true;
}
/**
* 获取未读消息数量
*/
public function getUnreadCount($userId) {
// 优先从缓存获取
if ($this->cache) {
$cached = $this->cache->get("unread_count:{$userId}");
if ($cached !== false) {
return (int)$cached;
}
}
$sql = "SELECT unread_count FROM user_message_counters WHERE user_id = ?";
$stmt = $this->pdo->prepare($sql);
$stmt->execute([$userId]);
$count = $stmt->fetchColumn();
// 缓存结果
if ($this->cache) {
$this->cache->set("unread_count:{$userId}", (int)$count, 300); // 5分钟缓存
}
return (int)$count;
}
/**
* 递增未读计数
*/
private function incrementUnreadCount($userId) {
$sql = "INSERT INTO user_message_counters (user_id, unread_count, last_message_id)
VALUES (?, 1, ?)
ON DUPLICATE KEY UPDATE
unread_count = unread_count + 1,
last_message_id = VALUES(last_message_id)";
$stmt = $this->pdo->prepare($sql);
$stmt->execute([$userId, $this->getLastMessageId($userId)]);
// 清除缓存
if ($this->cache) {
$this->cache->delete("unread_count:{$userId}");
}
}
/**
* 递减未读计数
*/
private function decrementUnreadCount($userId) {
$sql = "UPDATE user_message_counters
SET unread_count = GREATEST(unread_count - 1, 0)
WHERE user_id = ?";
$stmt = $this->pdo->prepare($sql);
$stmt->execute([$userId]);
// 清除缓存
if ($this->cache) {
$this->cache->delete("unread_count:{$userId}");
}
}
/**
* 获取最后一条消息ID
*/
private function getLastMessageId($userId) {
$sql = "SELECT id FROM messages
WHERE to_user_id = ?
ORDER BY id DESC LIMIT 1";
$stmt = $this->pdo->prepare($sql);
$stmt->execute([$userId]);
return $stmt->fetchColumn();
}
/**
* 推送实时通知
*/
private function pushNotification($userId) {
// 可以集成WebSocket、Redis发布订阅等
// 示例：使用Redis发布消息
if (class_exists('Redis')) {
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
$redis->publish("user:{$userId}:notifications", json_encode([
'type' => 'new_message',
'time' => date('Y-m-d H:i:s')
]));
}
}
/**
* 删除消息
*/
public function deleteMessage($userId, $messageId) {
$sql = "DELETE FROM message_status WHERE user_id = ? AND message_id = ?";
$stmt = $this->pdo->prepare($sql);
$stmt->execute([$userId, $messageId]);
// 检查是否需要更新计数
$this->recalculateUnreadCount($userId);
}
/**
* 重新计算未读计数（一致性修复）
*/
public function recalculateUnreadCount($userId) {
$sql = "SELECT COUNT(*) FROM message_status
WHERE user_id = ? AND is_read = 0";
$stmt = $this->pdo->prepare($sql);
$stmt->execute([$userId]);
$count = $stmt->fetchColumn();
$sql = "UPDATE user_message_counters SET unread_count = ? WHERE user_id = ?";
$stmt = $this->pdo->prepare($sql);
$stmt->execute([$count, $userId]);
if ($this->cache) {
$this->cache->set("unread_count:{$userId}", $count, 300);
}
}
/**
* 获取会话列表（最近消息）
*/
public function getConversationList($userId, $limit = 20) {
$sql = "SELECT
CASE
WHEN from_user_id = ? THEN to_user_id
ELSE from_user_id
END as peer_id,
u.username as peer_name,
m.content as last_message,
m.created_at as last_time,
COUNT(CASE WHEN ms.is_read = 0 AND m.from_user_id != ? THEN 1 END) as unread_count
FROM messages m
INNER JOIN users u ON CASE
WHEN from_user_id = ? THEN to_user_id
ELSE from_user_id
END = u.id
LEFT JOIN message_status ms ON m.id = ms.message_id AND ms.user_id = ?
WHERE m.to_user_id = ? OR m.from_user_id = ?
GROUP BY peer_id
ORDER BY m.created_at DESC
LIMIT ?";
$stmt = $this->pdo->prepare($sql);
$stmt->execute([$userId, $userId, $userId, $userId, $userId, $userId, $limit]);
return $stmt->fetchAll(PDO::FETCH_ASSOC);
}
}
API接口实现
api.php
<?php
class MessageAPI {
private $messageService;
private $user;
public function __construct(MessageService $messageService) {
$this->messageService = $messageService;
// 验证用户身份
$this->authenticate();
}
/**
* 获取未读消息数
*/
public function getUnreadCount() {
$count = $this->messageService->getUnreadCount($this->user['id']);
return $this->successResponse(['count' => $count]);
}
/**
* 获取未读消息列表
*/
public function getUnreadMessages() {
$userId = $this->user['id'];
$limit = isset($_GET['limit']) ? (int)$_GET['limit'] : 20;
$offset = isset($_GET['offset']) ? (int)$_GET['offset'] : 0;
$messages = $this->messageService->getUnreadMessages($userId, $limit, $offset);
return $this->successResponse(['messages' => $messages]);
}
/**
* 发送消息
*/
public function sendMessage() {
$input = json_decode(file_get_contents('php://input'), true);
if (!isset($input['to_user_id']) || !isset($input['content'])) {
return $this->errorResponse('参数错误');
}
$messageId = $this->messageService->sendMessage(
$this->user['id'],
$input['to_user_id'],
$input['content'],
isset($input['message_type']) ? $input['message_type'] : 1
);
return $this->successResponse(['message_id' => $messageId]);
}
/**
* 标记已读
*/
public function markAsRead() {
$input = json_decode(file_get_contents('php://input'), true);
if (!isset($input['message_id'])) {
return $this->errorResponse('缺少消息ID');
}
$result = $this->messageService->markAsRead($this->user['id'], $input['message_id']);
return $this->successResponse(['read' => $result]);
}
/**
* 获取会话列表
*/
public function getConversations() {
$conversations = $this->messageService->getConversationList($this->user['id']);
return $this->successResponse(['conversations' => $conversations]);
}
private function authenticate() {
// 实现用户认证逻辑
// 这里简化处理
$this->user = ['id' => 1]; // 示例用户ID
}
private function successResponse($data) {
header('Content-Type: application/json');
echo json_encode(['status' => 'success', 'data' => $data]);
exit;
}
private function errorResponse($message) {
header('Content-Type: application/json');
echo json_encode(['status' => 'error', 'message' => $message]);
exit;
}
}
前端JavaScript示例
messageNotification.js
class MessageNotification {
constructor(userId) {
this.userId = userId;
this.unreadCount = 0;
this.initialize();
}
initialize() {
// 首次加载获取未读数
this.fetchUnreadCount();
// 建立WebSocket连接（可选）
this.initWebSocket();
// 定时轮询（备选方案）
setInterval(() => this.fetchUnreadCount(), 30000);
}
async fetchUnreadCount() {
try {
const response = await fetch('/api/get_unread_count', {
headers: {
'Authorization': 'Bearer ' + localStorage.getItem('token')
}
});
const data = await response.json();
if (data.status === 'success') {
this.updateBadge(data.data.count);
}
} catch (error) {
console.error('获取未读消息失败:', error);
}
}
updateBadge(count) {
this.unreadCount = count;
const badge = document.getElementById('message-badge');
if (count > 0) {
badge.style.display = 'block';
badge.textContent = count > 99 ? '99+' : count;
} else {
badge.style.display = 'none';
}
// 更新页面标题
document.title = count > 0 ? `(${count}) 新消息 - ${document.title}` : document.title;
}
initWebSocket() {
// 使用WebSocket实现实时通知
const ws = new WebSocket(`ws://your-server.com/ws?userId=${this.userId}`);
ws.onmessage = (event) => {
const data = JSON.parse(event.data);
if (data.type === 'new_message') {
this.fetchUnreadCount();
this.showNotification(data);
}
};
}
showNotification(data) {
if (Notification.permission === 'granted') {
new Notification('新消息', {
body: data.content || '您有一条新消息',
icon: '/path/to/icon.png'
});
}
}
}
// 初始化
document.addEventListener('DOMContentLoaded', () => {
const userId = document.body.dataset.userId;
const notifier = new MessageNotification(userId);
});
优化建议
性能优化
- 使用Redis缓存未读计数，减少数据库查询
- 批量处理标记已读操作
- 分页加载消息列表，避免一次加载过多数据
扩展功能
- 消息推送集成WebSocket或第三方推送服务
- 多端同步支持Web、移动端、桌面端消息同步
- 消息撤回当消息被撤回时，通知相关用户
安全性考虑
- 过滤防止XSS攻击
- 访问控制验证用户权限
- 敏感信息加密
数据一致性
- 定期任务同步消息状态
- 异常监控检测未读计数不一致
- 重试机制处理消息发送失败
这个设计提供了完整、可扩展的未读消息系统解决方案,可以根据实际需求进行适当调整。
