本文目录导读：
我来详细介绍如何用PHP搭建IM即时通讯后端系统：
技术选型方案
核心组件
- WebSocket服务器: Swoole / Workerman
- 消息队列: Redis (Pub/Sub)
- 数据库: MySQL + Redis
- 协议支持: WebSocket + HTTP推荐框架
// 1. Workerman (更易上手) composer require workerman/workerman // 2. Swoole (性能更强) composer require swoole/swoole // 3. 集成方案 - TPIM composer require topthink/think-im
基础架构设计
1 完整目录结构
im-system/
├── server/
│ ├── websocket.php # WebSocket服务
│ ├── http.php # HTTP API服务
│ └── worker.php # 消息处理worker
├── app/
│ ├── controller/
│ │ ├── Auth.php # 认证
│ │ ├── Message.php # 消息处理
│ │ └── Friend.php # 好友关系
│ ├── model/
│ │ ├── User.php
│ │ ├── Message.php
│ │ └── Friend.php
│ ├── service/
│ │ ├── RedisService.php
│ │ ├── MessageService.php
│ │ └── UserService.php
│ └── helper/
│ └── function.php
├── config/
│ ├── app.php
│ └── database.php
├── vendor/
├── composer.json
└── start.php核心实现代码
1 WebSocket服务端 (Workerman示例)
server/websocket.php
<?php
require_once __DIR__ . '/../vendor/autoload.php';
use Workerman\Worker;
use Workerman\Lib\Timer;
// 创建WebSocket服务器
$ws_worker = new Worker('websocket://0.0.0.0:8282');
$ws_worker->count = 4; // 进程数
// 全局用户连接映射表
$userConnections = [];
$ws_worker->onConnect = function($connection) {
echo "New connection: {$connection->id}\n";
};
$ws_worker->onMessage = function($connection, $data) {
global $userConnections;
$message = json_decode($data, true);
if (!$message) return;
switch ($message['type']) {
case 'auth': // 用户认证
handleAuth($connection, $message, $userConnections);
break;
case 'chat': // 发送消息
handleChat($connection, $message, $userConnections);
break;
case 'heartbeat': // 心跳
$connection->send(json_encode(['type' => 'heartbeat_ack']));
break;
case 'offline': // 离线消息获取
getOfflineMessages($connection, $message);
break;
}
};
$ws_worker->onClose = function($connection) {
global $userConnections;
// 移除用户连接
foreach ($userConnections as $uid => $conn) {
if ($conn->id === $connection->id) {
unset($userConnections[$uid]);
// 广播离线状态
broadcastStatus($uid, 'offline');
break;
}
}
echo "Connection {$connection->id} closed\n";
};
// 用户认证处理
function handleAuth($connection, $message, &$userConnections) {
$token = $message['token'] ?? '';
$userId = validateToken($token);
if ($userId) {
$userConnections[$userId] = $connection;
$connection->uid = $userId;
// 发送认证成功响应
$connection->send(json_encode([
'type' => 'auth_success',
'user_id' => $userId
]));
// 同步离线消息
syncOfflineMessages($userId, $connection);
// 广播在线状态
broadcastStatus($userId, 'online');
} else {
$connection->send(json_encode([
'type' => 'auth_failed',
'message' => 'Invalid token'
]));
$connection->close();
}
}
// 处理聊天消息
function handleChat($connection, $message, $userConnections) {
$fromUid = $connection->uid;
$toUid = $message['to_user_id'];
$content = $message['content'];
$msgType = $message['msg_type'] ?? 'text';
// 生成消息ID和时间戳
$msgId = uniqid('msg_', true);
$timestamp = time();
// 保存消息到数据库
$msgData = [
'msg_id' => $msgId,
'from_uid' => $fromUid,
'to_uid' => $toUid,
'content' => $content,
'msg_type' => $msgType,
'timestamp' => $timestamp,
'status' => 0 // 未读
];
saveMessage($msgData);
// 判断目标是否在线
if (isset($userConnections[$toUid])) {
// 在线，直接发送
$userConnections[$toUid]->send(json_encode([
'type' => 'new_message',
'data' => $msgData
]));
// 更新消息状态为已送达
updateMessageStatus($msgId, 1);
// 发送送达回执
$connection->send(json_encode([
'type' => 'message_delivered',
'msg_id' => $msgId,
'timestamp' => $timestamp
]));
} else {
// 离线，消息已保存到数据库
$connection->send(json_encode([
'type' => 'message_stored',
'msg_id' => $msgId,
'timestamp' => $timestamp
]));
}
}
// 启动服务
Worker::runAll();
2 REST API服务 (消息管理)
server/http.php
<?php
require_once __DIR__ . '/../vendor/autoload.php';
use Workerman\Worker;
use Workerman\Protocols\Http;
$http_worker = new Worker('http://0.0.0.0:8280');
$http_worker->count = 4;
$http_worker->onMessage = function($connection, $request) {
$router = $request->path();
$method = $request->method();
// CORS头
$connection->sendHeader('Access-Control-Allow-Origin', '*');
$connection->sendHeader('Access-Control-Allow-Methods', 'GET, POST, PUT, DELETE, OPTIONS');
$connection->sendHeader('Access-Control-Allow-Headers', 'Content-Type, Authorization');
if ($method === 'OPTIONS') {
$connection->send('');
return;
}
$response = [];
switch (true) {
case $router === '/api/login' && $method === 'POST':
$response = handleLogin($request);
break;
case $router === '/api/register' && $method === 'POST':
$response = handleRegister($request);
break;
case $router === '/api/messages/history' && $method === 'GET':
$response = getMessageHistory($request);
break;
case $router === '/api/messages/offline' && $method === 'GET':
$response = getOfflineMessages($request);
break;
case preg_match('/\/api\/users\/(\d+)/', $router, $matches) && $method === 'GET':
$response = getUserInfo($matches[1]);
break;
default:
$response = ['code' => 404, 'message' => 'Not Found'];
}
$connection->send(json_encode($response));
};
// 登录处理
function handleLogin($request) {
$data = json_decode($request->rawBody(), true);
$username = $data['username'] ?? '';
$password = $data['password'] ?? '';
// 验证用户
$user = verifyUser($username, $password);
if ($user) {
// 生成token
$token = generateToken($user['id']);
// 返回WebSocket连接信息
return [
'code' => 200,
'data' => [
'token' => $token,
'user_id' => $user['id'],
'username' => $user['username'],
'avatar' => $user['avatar'],
'ws_url' => 'ws://127.0.0.1:8282'
]
];
}
return ['code' => 401, 'message' => 'Login failed'];
}
Worker::runAll();
3 消息服务模块
app/service/MessageService.php
<?php
namespace app\service;
use Redis;
use PDO;
class MessageService {
private $redis;
private $db;
public function __construct() {
$this->redis = new Redis();
$this->redis->connect('127.0.0.1', 6379);
$this->db = new PDO('mysql:host=localhost;dbname=im_db', 'root', 'password');
}
// 保存消息
public function saveMessage($msgData) {
// 保存到MySQL
$sql = "INSERT INTO messages (msg_id, from_uid, to_uid, content, msg_type, timestamp, status)
VALUES (?, ?, ?, ?, ?, ?, ?)";
$stmt = $this->db->prepare($sql);
$stmt->execute([
$msgData['msg_id'],
$msgData['from_uid'],
$msgData['to_uid'],
$msgData['content'],
$msgData['msg_type'],
$msgData['timestamp'],
$msgData['status']
]);
// 保存到Redis (最近消息缓存)
$cacheKey = "recent_messages:{$msgData['from_uid']}:{$msgData['to_uid']}";
$this->redis->lPush($cacheKey, json_encode($msgData));
$this->redis->lTrim($cacheKey, 0, 99); // 只保留最近100条
// 如果是离线消息，保存到离线队列
$this->addOfflineMessage($msgData['to_uid'], $msgData);
}
// 添加离线消息
public function addOfflineMessage($userId, $msgData) {
$key = "offline_messages:{$userId}";
$this->redis->rPush($key, json_encode($msgData));
$this->redis->expire($key, 86400); // 24小时过期
}
// 获取离线消息
public function getOfflineMessages($userId) {
$key = "offline_messages:{$userId}";
$messages = [];
while ($msg = $this->redis->lPop($key)) {
$messages[] = json_decode($msg, true);
}
return $messages;
}
// 获取聊天历史
public function getChatHistory($userId, $friendId, $limit = 20, $offset = 0) {
// 先从Redis获取缓存
$cacheKey = "recent_messages:{$userId}:{$friendId}";
$cached = $this->redis->lRange($cacheKey, $offset, $offset + $limit - 1);
if (!empty($cached)) {
return array_map('json_decode', $cached);
}
// Redis无缓存，从MySQL获取
$sql = "SELECT * FROM messages
WHERE (from_uid = ? AND to_uid = ?) OR (from_uid = ? AND to_uid = ?)
ORDER BY timestamp DESC
LIMIT ? OFFSET ?";
$stmt = $this->db->prepare($sql);
$stmt->execute([$userId, $friendId, $friendId, $userId, $limit, $offset]);
return $stmt->fetchAll(PDO::FETCH_ASSOC);
}
// 更新消息状态
public function updateMessageStatus($msgId, $status) {
$sql = "UPDATE messages SET status = ? WHERE msg_id = ?";
$stmt = $this->db->prepare($sql);
return $stmt->execute([$status, $msgId]);
}
}
数据库设计
1 MySQL表结构
-- 用户表
CREATE TABLE `users` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`username` varchar(50) NOT NULL,
`password` varchar(255) NOT NULL,
`avatar` varchar(255) DEFAULT NULL,
`status` tinyint(1) DEFAULT '0' COMMENT '0离线 1在线',
`last_login` timestamp NULL,
`created_at` timestamp DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
UNIQUE KEY `username` (`username`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
-- 消息表
CREATE TABLE `messages` (
`id` bigint(20) NOT NULL AUTO_INCREMENT,
`msg_id` varchar(50) NOT NULL,
`from_uid` int(11) NOT NULL,
`to_uid` int(11) NOT NULL,
`content` text NOT NULL,
`msg_type` varchar(20) DEFAULT 'text' COMMENT 'text/image/file',
`timestamp` int(11) NOT NULL,
`status` tinyint(1) DEFAULT '0' COMMENT '0未读 1已送达 2已读',
`created_at` timestamp DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
INDEX `idx_from_to` (`from_uid`, `to_uid`),
INDEX `idx_msg_id` (`msg_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
-- 好友关系表
CREATE TABLE `friends` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`user_id` int(11) NOT NULL,
`friend_id` int(11) NOT NULL,
`status` tinyint(1) DEFAULT '0' COMMENT '0待确认 1已确认',
`created_at` timestamp DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
UNIQUE KEY `uk_user_friend` (`user_id`, `friend_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
2 Redis数据结构
// 用户在线状态
$redis->hSet('online_users', $userId, $connectionId);
$redis->hDel('online_users', $userId);
// 用户会话信息
$redis->set("user_session:{$userId}", json_encode([
'token' => $token,
'expire' => time() + 86400
]));
$redis->expire("user_session:{$userId}", 86400);
// 离线消息队列
$redis->rPush("offline_messages:{$userId}", json_encode($msgData));
$redis->expire("offline_messages:{$userId}", 86400);
// 最近聊天缓存
$redis->lPush("recent_chat:{$userId}:{$friendId}", json_encode($msgData));
$redis->lTrim("recent_chat:{$userId}:{$friendId}", 0, 99);
启动服务
1 服务启动脚本
start.php
<?php
require_once __DIR__ . '/vendor/autoload.php';
// 启动WebSocket服务
$wsProcess = popen('php server/websocket.php start -d', 'r');
// 启动HTTP服务
$httpProcess = popen('php server/http.php start -d', 'r');
echo "IM Server started successfully!\n";
echo "WebSocket: ws://0.0.0.0:8282\n";
echo "HTTP API: http://0.0.0.0:8280\n";
2 生产环境部署
# 使用Supervisor管理进程 sudo apt-get install supervisor # 配置文件 /etc/supervisor/conf.d/im-server.conf [program:im-websocket] command=php /path/to/server/websocket.php start numprocs=1 autostart=true autorestart=true user=www-data [program:im-http] command=php /path/to/server/http.php start numprocs=1 autostart=true autorestart=true user=www-data # 启动 sudo supervisorctl reload sudo supervisorctl start all
前端JS连接示例
// 连接WebSocket
const ws = new WebSocket('ws://127.0.0.1:8282');
// 认证
ws.onopen = function() {
ws.send(JSON.stringify({
type: 'auth',
token: 'your_token_here'
}));
};
// 接收消息
ws.onmessage = function(event) {
const data = JSON.parse(event.data);
switch(data.type) {
case 'new_message':
// 处理新消息
showMessage(data.data);
break;
case 'auth_success':
console.log('连接成功');
break;
case 'heartbeat_ack':
// 心跳响应
break;
}
};
// 发送消息
function sendMessage(toUserId, content) {
ws.send(JSON.stringify({
type: 'chat',
to_user_id: toUserId,
content: content,
msg_type: 'text'
}));
}
// 心跳保持
setInterval(() => {
ws.send(JSON.stringify({ type: 'heartbeat' }));
}, 30000);
性能优化建议
1 消息合并发送
// 批量处理消息
function batchSendMessages() {
$batchSize = 100;
$messages = [];
// 收集消息
while (count($messages) < $batchSize) {
$msg = getNextMessage();
if (!$msg) break;
$messages[] = $msg;
}
// 批量发送
foreach ($messages as $msg) {
sendToUser($msg['to_uid'], $msg);
}
}
2 连接池优化
// 使用Swoole连接池
$pool = new Swoole\Database\PDOPool(
new Swoole\Database\PDOConfig()
);
// 从池中获取连接
$pdo = $pool->get();
$stmt = $pdo->prepare('SELECT * FROM users WHERE id = ?');
$stmt->execute([$userId]);
$pool->put($pdo);
3 消息压缩
// 大消息压缩
function compressMessage($data) {
if (strlen($data) > 1024) {
return gzcompress($data, 6);
}
return $data;
}
这个PHP IM系统实现了：
- ✅ WebSocket实时通信
- ✅ 用户认证系统
- ✅ 消息存储和转发
- ✅ 离线消息处理
- ✅ 消息历史查询
- ✅ 心跳检测机制
- ✅ 分布式部署支持
可以根据实际需求扩展群聊、文件传输、消息回执等功能，对于高并发场景,建议使用Swoole替代Workerman以获得更好的性能。
