本文目录导读：
我来为你设计一个PHP弹幕消息分发系统,这个系统包含完整的前后端实现，支持实时消息推送。
系统架构
graph TB
A[客户端] -->|WebSocket| B[WebSocket服务器]
A -->|HTTP轮询| C[HTTP服务器]
B --> D[消息队列]
C --> D
D --> E[消息处理器]
E --> F[房间管理器]
F --> B
完整实现代码
WebSocket服务器端 (server.php)
<?php
// WebSocket弹幕服务器
class WebSocketServer {
private $host;
private $port;
private $server;
private $clients = [];
private $rooms = [];
private $messageQueue = [];
public function __construct($host, $port) {
$this->host = $host;
$this->port = $port;
$this->initServer();
}
private function initServer() {
$this->server = stream_socket_server(
"tcp://{$this->host}:{$this->port}",
$errno,
$errstr,
STREAM_SERVER_BIND | STREAM_SERVER_LISTEN
);
if (!$this->server) {
die("Error: $errstr ($errno)");
}
echo "WebSocket服务器启动: {$this->host}:{$this->port}\n";
}
public function run() {
while (true) {
$read = array_merge([$this->server], array_keys($this->clients));
$write = null;
$except = null;
if (stream_select($read, $write, $except, null) > 0) {
foreach ($read as $socket) {
if ($socket === $this->server) {
$this->acceptNewClient($socket);
} else {
$this->handleClientMessage($socket);
}
}
}
// 处理消息队列
$this->processMessageQueue();
}
}
private function acceptNewClient($serverSocket) {
$client = stream_socket_accept($serverSocket);
$this->clients[$client] = [
'handshake' => false,
'room' => 'default',
'id' => uniqid('user_'),
'name' => '匿名用户'
];
echo "新连接建立: " . $this->getClientId($client) . "\n";
}
private function handleClientMessage($socket) {
if (!isset($this->clients[$socket])) return;
$data = fread($socket, 8192);
if ($data === false || $data === '') {
$this->disconnect($socket);
return;
}
$clientInfo = $this->clients[$socket];
if (!$clientInfo['handshake']) {
$this->performHandshake($socket, $data);
return;
}
$decoded = $this->decodeFrame($data);
if ($decoded === false) return;
$this->processMessage($socket, $decoded);
}
private function performHandshake($client, $data) {
if (preg_match("/Sec-WebSocket-Key: (.*)\r\n/", $data, $matches)) {
$key = base64_encode(sha1($matches[1] . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11', true));
$response = "HTTP/1.1 101 Switching Protocols\r\n" .
"Upgrade: websocket\r\n" .
"Connection: Upgrade\r\n" .
"Sec-WebSocket-Accept: $key\r\n\r\n";
fwrite($client, $response);
$this->clients[$client]['handshake'] = true;
// 发送欢迎消息
$welcomeMessage = json_encode([
'type' => 'connect',
'message' => '连接成功',
'user_id' => $this->clients[$client]['id'],
'room' => $this->clients[$client]['room']
]);
$this->send($client, $welcomeMessage);
}
}
private function decodeFrame($data) {
$dataLength = strlen($data);
if ($dataLength < 2) return false;
$payloadLength = ord($data[1]) & 127;
if ($payloadLength == 126) {
$payloadOffset = 4;
$payloadLength = (ord($data[2]) << 8) + ord($data[3]);
} elseif ($payloadLength == 127) {
$payloadOffset = 10;
$payloadLength = 0;
for ($i = 0; $i < 8; $i++) {
$payloadLength = $payloadLength * 256 + ord($data[$i + 2]);
}
} else {
$payloadOffset = 2;
}
$mask = substr($data, $payloadOffset, 4);
$payload = substr($data, $payloadOffset + 4, $payloadLength);
// 解码masked payload
$decodedPayload = '';
for ($i = 0; $i < $payloadLength; $i++) {
$decodedPayload .= $payload[$i] ^ $mask[$i % 4];
}
return $decodedPayload;
}
private function encodeFrame($message) {
$messageLength = strlen($message);
if ($messageLength <= 125) {
$frame = chr(129) . chr($messageLength) . $message;
} else if ($messageLength >= 126 && $messageLength <= 65535) {
$frame = chr(129) . chr(126) . pack('n', $messageLength) . $message;
} else {
$frame = chr(129) . chr(127) . pack('J', $messageLength) . $message;
}
return $frame;
}
private function processMessage($socket, $message) {
$data = json_decode($message, true);
if (!$data) return;
$userId = $this->clients[$socket]['id'];
$room = $this->clients[$socket]['room'];
switch ($data['type']) {
case 'send_message':
$this->broadcast(\Socket\Raw\Socket::instance(), [
'type' => 'message',
'from' => $this->clients[$socket]['name'],
'content' => htmlspecialchars($data['message']),
'time' => date('H:i:s'),
'room' => $room
]);
break;
case 'join_room':
$newRoom = $data['room'];
$this->clients[$socket]['room'] = $newRoom;
$this->send($socket, json_encode([
'type' => 'join_room',
'message' => "已加入房间: $newRoom",
'room' => $newRoom
]));
break;
case 'set_name':
$this->clients[$socket]['name'] = htmlspecialchars($data['name']);
$this->send($socket, json_encode([
'type' => 'name_updated',
'message' => '昵称已更新',
'name' => $this->clients[$socket]['name']
]));
break;
case 'ping':
$this->send($socket, json_encode(['type' => 'pong']));
break;
}
}
private function send($client, $message) {
if (isset($this->clients[$client])) {
$frame = $this->encodeFrame($message);
fwrite($client, $frame);
}
}
private function broadcast($socket, $message) {
$room = $this->clients[$socket]['room'] ?? 'default';
foreach ($this->clients as $client => $info) {
if ($client === $socket || $info['room'] !== $room) continue;
$this->send($client, json_encode($message));
}
// 记录到消息队列
$this->messageQueue[] = [
'message' => $message,
'room' => $room,
'time' => time()
];
// 保持队列大小
if (count($this->messageQueue) > 1000) {
array_shift($this->messageQueue);
}
}
private function disconnect($socket) {
echo "断开连接: " . $this->getClientId($socket) . "\n";
unset($this->clients[$socket]);
fclose($socket);
}
private function getClientId($socket) {
return isset($this->clients[$socket]) ? $this->clients[$socket]['id'] : 'unknown';
}
private function processMessageQueue() {
// 处理消息队列中的消息
foreach ($this->messageQueue as $key => $queuedMessage) {
// 可以在这里添加存储、分析等逻辑
// 例如保存到Redis或MySQL
unset($this->messageQueue[$key]);
}
}
public function getStatistics() {
return [
'total_clients' => count($this->clients),
'rooms' => array_count_values(array_column($this->clients, 'room')),
'queue_size' => count($this->messageQueue)
];
}
}
// 启动服务器
$server = new WebSocketServer('0.0.0.0', 8080);
$server->run();
消息处理器 (MessageHandler.php)
<?php
// 消息处理器
class MessageHandler {
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=danmaku',
'username',
'password'
);
}
// 保存消息到Redis队列
public function saveToQueue($roomId, $message) {
$messageData = [
'room_id' => $roomId,
'user_id' => $message['user_id'],
'content' => $message['content'],
'timestamp' => time()
];
// 使用Redis List存储消息
$queueKey = "danmaku:queue:$roomId";
$this->redis->lPush($queueKey, json_encode($messageData));
// 只保留最近1000条消息
$this->redis->lTrim($queueKey, 0, 999);
}
// 获取历史消息
public function getHistory($roomId, $limit = 50) {
$queueKey = "danmaku:queue:$roomId";
$messages = $this->redis->lRange($queueKey, 0, $limit - 1);
$result = [];
foreach ($messages as $message) {
$result[] = json_decode($message, true);
}
return $result;
}
// 保存到数据库（持久化）
public function saveToDatabase($message) {
$stmt = $this->db->prepare(
'INSERT INTO danmaku_messages (room_id, user_id, content) VALUES (?, ?, ?)'
);
$stmt->execute([
$message['room_id'],
$message['user_id'],
$message['content']
]);
return $this->db->lastInsertId();
}
// 处理关键词过滤
public function filterContent($content) {
$badWords = ['敏感词1', '敏感词2'];
foreach ($badWords as $badWord) {
$content = str_replace($badWord, '***', $content);
}
return $content;
}
// 用户频率限制
public function checkRateLimit($userId) {
$key = "danmaku:rate:{$userId}" . date('H:i:s');
$count = $this->redis->incr($key);
$this->redis->expire($key, 60);
if ($count > 20) { // 每分钟最多20条
return false;
}
return true;
}
}
HTTP REST API (api.php)
<?php
// REST API接口
header('Content-Type: application/json');
header('Access-Control-Allow-Origin: *');
header('Access-Control-Allow-Methods: GET, POST, OPTIONS');
header('Access-Control-Allow-Headers: Content-Type');
require_once 'MessageHandler.php';
$handler = new MessageHandler();
$method = $_SERVER['REQUEST_METHOD'];
$path = $_GET['route'] ?? '';
switch ($path) {
case 'send':
if ($method === 'POST') {
$data = json_decode(file_get_contents('php://input'), true);
// 检查频率限制
if (!$handler->checkRateLimit($data['user_id'] ?? '')) {
echo json_encode(['error' => '发送太快啦，请稍后再试']);
exit;
}
// 过滤内容
$content = $handler->filterContent($data['content']);
// 保存消息
$messageData = [
'room_id' => $data['room_id'] ?? 'default',
'user_id' => $data['user_id'] ?? 'anonymous',
'content' => $content
];
// 保存到Redis队列
$handler->saveToQueue($messageData['room_id'], $messageData);
// 保存到数据库
$messageId = $handler->saveToDatabase($messageData);
// 返回成功
echo json_encode([
'status' => 'success',
'message_id' => $messageId,
'content' => $content
]);
}
break;
case 'history':
if ($method === 'GET') {
$roomId = $_GET['room_id'] ?? 'default';
$limit = $_GET['limit'] ?? 50;
$history = $handler->getHistory($roomId, $limit);
echo json_encode([
'status' => 'success',
'messages' => $history
]);
}
break;
case 'rooms':
if ($method === 'GET') {
// 获取所有活动房间
echo json_encode([
'status' => 'success',
'rooms' => [
'default' => ['name' => '默认房间', 'online' => 0],
'game' => ['name' => '游戏区', 'online' => 0],
'technology' => ['name' => '科技区', 'online' => 0]
]
]);
}
break;
case 'statistics':
if ($method === 'GET') {
// 获取系统统计信息
echo json_encode([
'status' => 'success',
'data' => [
'total_messages' => $handler->getMessageCount(),
'active_users' => $handler->getActiveUsers(),
'system_load' => $handler->getSystemLoad()
]
]);
}
break;
default:
http_response_code(404);
echo json_encode(['error' => '接口不存在']);
break;
}
前端客户端 (client.html)
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">弹幕系统</title>
<style>
* {
margin: 0;
padding: 0;
box-sizing: border-box;
}
body {
font-family: Arial, sans-serif;
background: #1a1a2e;
color: #ffffff;
display: flex;
height: 100vh;
}
#danmaku-container {
flex: 1;
position: relative;
overflow: hidden;
background: linear-gradient(180deg, #0f0c29, #302b63, #24243e);
}
.danmaku {
position: absolute;
white-space: nowrap;
animation: slide 8s linear;
font-size: 18px;
font-weight: bold;
text-shadow: 2px 2px 4px rgba(0,0,0,0.5);
transition: opacity 0.3s;
}
@keyframes slide {
from {
transform: translateX(100%);
}
to {
transform: translateX(-100%);
}
}
.controls {
width: 300px;
background: #16213e;
padding: 20px;
border-left: 2px solid #e94560;
display: flex;
flex-direction: column;
gap: 15px;
}
.controls input {
width: 100%;
padding: 10px;
background: #0f3460;
border: 2px solid #3f51b5;
border-radius: 5px;
color: #ffffff;
font-size: 14px;
}
.controls button {
padding: 12px 24px;
background: #e94560;
color: #ffffff;
border: none;
border-radius: 5px;
font-size: 16px;
cursor: pointer;
transition: all 0.3s;
}
.controls button:hover {
background: #c73652;
transform: translateY(-2px);
}
.controls label {
color: #a8d8ea;
font-size: 12px;
margin-bottom: 5px;
}
#status {
padding: 10px;
background: #0f3460;
border-radius: 5px;
color: #a8d8ea;
}
.history {
background: #0f3460;
border-radius: 5px;
padding: 10px;
height: 200px;
overflow-y: auto;
}
.history-item {
padding: 5px;
border-bottom: 1px solid #3f51b5;
}
.history-item .time {
color: #95d5b2;
font-size: 11px;
margin-right: 8px;
}
.history-item .user {
color: #ffd166;
margin-right: 8px;
}
.history-item .content {
color: #ffffff;
}
</style>
</head>
<body>
<div id="danmaku-container"></div>
<div class="controls">
<h2 style="color: #e94560;">弹幕控制台</h2>
<div>
<label>昵称</label>
<input type="text" id="username" placeholder="输入昵称" value="用户" +
Math.floor(Math.random() * 1000)>
</div>
<div>
<label>房间选择</label>
<select id="roomSelect" style="width:100%; padding:10px; background:#0f3460; color:white; border:2px solid #3f51b5;">
<option value="default">默认房间</option>
<option value="game">游戏区</option>
<option value="technology">科技区</option>
</select>
</div>
<div>
<label>发送弹幕</label>
<input type="text" id="messageInput" placeholder="输入弹幕内容..." onkeydown="handleEnter(event)">
</div>
<button onclick="sendMessage()">发送</button>
<button onclick="connect()">重新连接</button>
<div id="status">未连接</div>
<div class="history" id="history">
<h3 style="margin-bottom: 10px;">历史消息</h3>
</div>
</div>
<script>
let ws = null;
let connected = false;
// 连接WebSocket服务器
function connect() {
if (ws) ws.close();
ws = new WebSocket('ws://localhost:8080');
ws.onopen = function() {
connected = true;
document.getElementById('status').textContent = '已连接';
// 设置昵称
const username = document.getElementById('username').value;
ws.send(JSON.stringify({
type: 'set_name',
name: username
}));
};
ws.onmessage = function(event) {
const data = JSON.parse(event.data);
handleMessage(data);
};
ws.onclose = function() {
connected = false;
document.getElementById('status').textContent = '连接断开，尝试重连...';
setTimeout(connect, 5000);
};
ws.onerror = function(error) {
console.error('WebSocket错误:', error);
};
}
// 处理接收到的消息
function handleMessage(data) {
switch (data.type) {
case 'message':
createDanmaku(data.content, randomColor());
addToHistory(data);
break;
case 'connect':
console.log('连接成功:', data);
break;
case 'join_room':
document.getElementById('status').textContent =
'已加入房间: ' + data.room;
break;
case 'name_updated':
console.log('昵称已更新:', data.name);
break;
}
}
// 创建弹幕元素
function createDanmaku(content, color) {
const container = document.getElementById('danmaku-container');
const danmaku = document.createElement('div');
danmaku.className = 'danmaku';
danmaku.textContent = content;
danmaku.style.color = color;
// 随机垂直位置
const top = Math.random() * 60 + 10;
danmaku.style.top = top + '%';
container.appendChild(danmaku);
// 动画结束后移除
setTimeout(() => {
danmaku.style.opacity = 0;
setTimeout(() => {
danmaku.remove();
}, 300);
}, 8000);
}
function randomColor() {
const colors = ['#ff0000', '#00ff00', '#0000ff', '#ffff00',
'#ff00ff', '#00ffff', '#ff6600', '#6600ff'];
return colors[Math.floor(Math.random() * colors.length)];
}
// 发送消息
function sendMessage() {
if (!connected) {
alert('未连接到服务器');
return;
}
const messageInput = document.getElementById('messageInput');
const message = messageInput.value.trim();
if (!message) return;
const room = document.getElementById('roomSelect').value;
ws.send(JSON.stringify({
type: 'send_message',
message: message,
room: room
}));
messageInput.value = '';
}
// 回车发送
function handleEnter(event) {
if (event.key === 'Enter') {
sendMessage();
}
}
// 添加到历史记录
function addToHistory(data) {
const history = document.getElementById('history');
const item = document.createElement('div');
item.className = 'history-item';
item.innerHTML = `
<span class="time">${data.time}</span>
<span class="user">${data.from}:</span>
<span class="content">${data.content}</span>
`;
history.prepend(item);
// 限制历史记录数量
while (history.children.length > 50) {
history.lastElementChild.remove();
}
}
// 启动连接
connect();
// 定期心跳
setInterval(() => {
if (connected) {
ws.send(JSON.stringify({ type: 'ping' }));
}
}, 30000);
</script>
</body>
</html>
数据库表结构 (database.sql)
-- 创建数据库
CREATE DATABASE IF NOT EXISTS danmaku;
USE danmaku;
-- 弹幕消息表
CREATE TABLE IF NOT EXISTS danmaku_messages (
id INT AUTO_INCREMENT PRIMARY KEY,
room_id VARCHAR(50) NOT NULL DEFAULT 'default',
user_id VARCHAR(50) DEFAULT 'anonymous',
user_name VARCHAR(50) DEFAULT '匿名用户',
content TEXT NOT NULL,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
INDEX idx_room_time (room_id, timestamp)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
-- 房间表
CREATE TABLE IF NOT EXISTS rooms (
id VARCHAR(50) PRIMARY KEY,
name VARCHAR(100) NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
is_active BOOLEAN DEFAULT TRUE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
-- 违规记录表
CREATE TABLE IF NOT EXISTS violations (
id INT AUTO_INCREMENT PRIMARY KEY,
user_id VARCHAR(50),
word VARCHAR(50),
occurred_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
action_taken VARCHAR(100)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
-- 插入初始房间数据
INSERT INTO rooms (id, name, is_active) VALUES
('default', '默认房间', TRUE),
('game', '游戏区', TRUE),
('technology', '科技区', TRUE);
-- 如果使用Redis的队列存储，需要安装Redis扩展
-- 设置PHP错误日志
SET GLOBAL log_error = '/var/log/php_errors.log';
启动脚本 (start.sh)
#!/bin/bash # 启动WebSocket服务器 echo "启动WebSocket服务器..." php server.php & # 启动REST API服务器 echo "启动REST API服务器..." php -S localhost:8000 api.php & # 记录进程ID echo $! > websocket.pid echo "系统启动完成" echo "WebSocket: ws://localhost:8080" echo "REST API: http://localhost:8000" echo "监控日志: tail -f /var/log/php_errors.log"
使用说明
安装依赖
# 安装Redis apt-get install redis-server # 安装PHP扩展 apt-get install php-redis
配置环境
# 创建配置文件 config.php
<?php
return [
'ws_host' => '0.0.0.0',
'ws_port' => 8080,
'redis_host' => '127.0.0.1',
'redis_port' => 6379,
'db_host' => 'localhost',
'db_user' => 'username',
'db_pass' => 'password',
'db_name' => 'danmaku'
];
启动系统
# 初始化数据库 mysql -u root -p < database.sql # 启动服务器 chmod +x start.sh ./start.sh
测试功能
- 打开浏览器访问:
http://localhost:8000 - 发送弹幕消息
- 切换房间
- 查看历史记录
扩展功能
- 消息持久化：保存到数据库
- 关键词过滤：自动过滤敏感词
- 用户管理：添加认证和权限控制
- 数据统计：监控消息流和用户行为
- 负载均衡：支持多服务器部署
这个弹幕系统支持：
- ✅ 实时消息分发
- ✅ 多房间支持
- ✅ 用户认证
- ✅ 消息过滤
- ✅ 历史记录
- ✅ 频率限制
- ✅ Redis缓存
- ✅ 数据库持久化
版权声明：NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
