PHP项目IM即时通讯如何PHP搭建后端

wen PHP项目 2026-06-29 220

本文目录导读：

技术选型方案
基础架构设计
核心实现代码
数据库设计
启动服务
前端JS连接示例
性能优化建议

我来详细介绍如何用PHP搭建IM即时通讯后端系统：

技术选型方案

核心组件

- WebSocket服务器: Swoole / Workerman
- 消息队列: Redis (Pub/Sub)
- 数据库: MySQL + Redis
- 协议支持: WebSocket + HTTP

基础架构设计

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以获得更好的性能。

本文地址： https://www.bmjup.com/post/11555.html

PHP项目IM即时通讯如何PHP搭建后端

技术选型方案

核心组件

推荐框架

基础架构设计

1 完整目录结构

核心实现代码

1 WebSocket服务端 (Workerman示例)

2 REST API服务 (消息管理)

3 消息服务模块

数据库设计

1 MySQL表结构

2 Redis数据结构

启动服务

1 服务启动脚本

2 生产环境部署

前端JS连接示例

性能优化建议

1 消息合并发送

2 连接池优化

3 消息压缩