php+最新的swoole 文件分发服务(类 FastDFS Tracker)
·
PHP 8.3 + Swoole 6.x 文件分发服务(类 FastDFS Tracker)· 完整实现
FastDFS 模型核心:Tracker(调度) + Storage(存储) + Group(副本组)。
本方案三个组件全部用 PHP+Swoole 实现,客户端拿到的 file_id 形如 group1/M00/00/AB/wKgABcd...jpg,与 FastDFS
协议兼容设计。
---
一、整体架构(大白话)
┌────────┐ ①upload? ┌──────────────┐
│ Client │ ───────────→│ Tracker │
│ │ ←node info │ (调度大脑) │
└────────┘ │ │
│ │ 维护: │
│ ②PUT file │ • groups │
▼ │ • storages │
┌────────────┐ │ • 心跳/健康 │
│ Storage 1 │ ←────────│• 负载均衡 │
│ (group1) │ replicate└──────────────┘
└──────┬─────┘ ▲
│ async │ heartbeat (5s)
▼ │
┌────────────┐ │
│ Storage 2 │ ───────────────┘
│ (group1) │
└────────────┘
Group 内多副本(强可用),Group 间数据不同(横向扩容)
核心思想:Tracker 只调度不存数据;Storage 只存数据自管副本;client 拿到 file_id 后下载可直连 Storage 跳过 Tracker。
---
二、最佳技术选型
┌───────────────────┬───────────────────────────────────────┬────────────────────────┐
│ 层 │ 库/技术 │ 原因 │
├───────────────────┼───────────────────────────────────────┼────────────────────────┤
│ 运行时 │ Swoole 6.x │ 协程 + sendfile 零拷贝 │
├───────────────────┼───────────────────────────────────────┼────────────────────────┤
│ 节点注册表 │ Swoole\Table │ 跨 worker 共享,纳秒级 │
├───────────────────┼───────────────────────────────────────┼────────────────────────┤
│ 元数据 │ MySQL + Redis 缓存 │ 持久 + 加速 │
├───────────────────┼───────────────────────────────────────┼────────────────────────┤
│ Tracker ↔Storage │ HTTP + JSON(也可 TCP 二进制) │ 简单 + 易调试 │
├───────────────────┼───────────────────────────────────────┼────────────────────────┤
│ 文件传输 │ Swoole sendfile + Range │ 零拷贝大文件 │
├───────────────────┼───────────────────────────────────────┼────────────────────────┤
│ 客户端 │ Swoole\Coroutine\Http\Client │ 协程化 │
├───────────────────┼───────────────────────────────────────┼────────────────────────┤
│ 副本同步 │ Swoole Task + 重试队列 │ 异步可靠 │
├───────────────────┼───────────────────────────────────────┼────────────────────────┤
│ 路由 │ nikic/fast-route │ 极快 │
├───────────────────┼───────────────────────────────────────┼────────────────────────┤
│ 哈希 │ CRC32 / 一致性哈希(可选) │ 路由稳定 │
├───────────────────┼───────────────────────────────────────┼────────────────────────┤
│ 监控 │ promphp/prometheus_client_php │ 标准指标 │
├───────────────────┼───────────────────────────────────────┼────────────────────────┤
│ 文件 ID │ 自研(group/M00/分目录/timestamp+rand) │ 兼容 FastDFS 风格 │
└───────────────────┴───────────────────────────────────────┴────────────────────────┘
composer require swoole/ide-helper nikic/fast-route promphp/prometheus_client_php monolog/monolog
---
三、文件 ID 格式
group1/M00/00/AB/wKgABcdef0123456.jpg
└──┬──┘└─┬┘└──┬──┘└──────┬───────┘
group 存储路径前缀 两级散列目录 文件名(base64时间戳+random)+扩展名
为什么这么设计:
- group1 →Tracker 调度到哪个副本组
- M00 →Storage 节点上挂载的第几块盘
- 00/AB →两级散列分目录,避免单目录文件过多导致 inode 慢
- 客户端拿到这个字符串就能直接下载,不再问 Tracker
---
四、完整代码
A. Tracker 服务(调度大脑)
1. 入口
<?php
// tracker.php
declare(strict_types=1);
require __DIR__ . '/vendor/autoload.php';
use Swoole\Http\Server;
use Swoole\Http\Request;
use Swoole\Http\Response;
use App\FDFS\Tracker\Container;
use App\FDFS\Tracker\Router;
\Swoole\Runtime::enableCoroutine(SWOOLE_HOOK_ALL);
$c = Container::boot();
$server = new Server('0.0.0.0', 22122, SWOOLE_PROCESS); // FastDFS 默认端口
$server->set([
'worker_num' => swoole_cpu_num(),
'task_worker_num' => 4,
'task_enable_coroutine' => true,
'enable_coroutine' => true,
'max_request' => 50000,
'hook_flags' => SWOOLE_HOOK_ALL,
'reactor_num' => swoole_cpu_num(),
]);
$server->on('WorkerStart', function($s, $wid) use ($c) {
$c->initPools();
$c->loadStoragesFromDB();
if ($wid === 0) {
// worker 0 跑健康检查
\Swoole\Coroutine::create(fn() => (new \App\FDFS\Tracker\HealthChecker($c))->run());
}
});
$server->on('Request', function(Request $req, Response $res) use ($c) {
(new Router($c))->dispatch($req, $res);
});
$server->start();
---
2. Container:节点注册表
<?php
// src/FDFS/Tracker/Container.php
namespace App\FDFS\Tracker;
use Swoole\Table;
use Swoole\Coroutine\Channel;
use Swoole\Coroutine\Redis;
use Swoole\Coroutine\MySQL;
class Container
{
public Table $storages; // node_id => 节点信息
public Table $groupStats; // group_name => 组统计(总容量/已用/活跃节点数)
public Channel $redisPool;
public Channel $mysqlPool;
public array $config;
public static function boot(): self
{
$c = new self();
$c->storages = new Table(1024);
$c->storages->column('node_id', Table::TYPE_STRING, 32);
$c->storages->column('group', Table::TYPE_STRING, 32);
$c->storages->column('host', Table::TYPE_STRING, 64);
$c->storages->column('port', Table::TYPE_INT, 4);
$c->storages->column('status', Table::TYPE_STRING, 16); // ACTIVE/SYNCING/OFFLINE
$c->storages->column('total_mb', Table::TYPE_INT, 8);
$c->storages->column('free_mb', Table::TYPE_INT, 8);
$c->storages->column('upload_qps',Table::TYPE_INT, 4);
$c->storages->column('last_hb', Table::TYPE_INT, 8);
$c->storages->column('weight', Table::TYPE_INT, 4);
$c->storages->create();
$c->groupStats = new Table(256);
$c->groupStats->column('total_nodes', Table::TYPE_INT, 4);
$c->groupStats->column('active_nodes', Table::TYPE_INT, 4);
$c->groupStats->column('total_free_mb',Table::TYPE_INT, 8);
$c->groupStats->create();
$c->config = [
'select_policy' => 'least_used', // round_robin / least_used / weighted / hash
'min_free_mb' => 1024, // 节点剩余 < 1GB 不再分配
'hb_timeout' => 15, // 心跳超过 15s →OFFLINE
];
return $c;
}
public function initPools(): void
{
$this->redisPool = new Channel(32);
for ($i=0;$i<32;$i++) {
$r = new Redis(); $r->connect('127.0.0.1', 6379);
$this->redisPool->push($r);
}
$this->mysqlPool = new Channel(8);
for ($i=0;$i<8;$i++) {
$db = new MySQL();
$db->connect(['host'=>'127.0.0.1','user'=>'root','password'=>'',
'database'=>'fdfs','charset'=>'utf8mb4']);
$this->mysqlPool->push($db);
}
}
public function loadStoragesFromDB(): void
{
$this->withMySQL(function($db) {
$rows = $db->query("SELECT * FROM storages WHERE enabled=1");
foreach ($rows as $r) {
$this->storages->set($r['node_id'], [
'node_id'=>$r['node_id'],
'group' =>$r['group_name'],
'host' =>$r['host'],
'port' =>(int)$r['port'],
'status' =>'OFFLINE', // 等心跳上来再标 ACTIVE
'total_mb'=>(int)$r['total_mb'],
'free_mb'=>0,
'upload_qps'=>0,
'last_hb'=>0,
'weight' =>(int)$r['weight'],
]);
}
});
$this->rebuildGroupStats();
}
public function rebuildGroupStats(): void
{
$stats = [];
foreach ($this->storages as $node) {
$g = $node['group'];
$stats[$g] ??= ['total_nodes'=>0,'active_nodes'=>0,'total_free_mb'=>0];
$stats[$g]['total_nodes']++;
if ($node['status'] === 'ACTIVE') {
$stats[$g]['active_nodes']++;
$stats[$g]['total_free_mb'] += $node['free_mb'];
}
}
foreach ($stats as $g => $s) $this->groupStats->set($g, $s);
}
public function withRedis(callable $fn) {
$r = $this->redisPool->pop();
try { return $fn($r); } finally { $this->redisPool->push($r); }
}
public function withMySQL(callable $fn) {
$db = $this->mysqlPool->pop();
try { return $fn($db); } finally { $this->mysqlPool->push($db); }
}
}
解释:
- storages 表是调度核心:每条记录一个 Storage 节点状态
- groupStats 是聚合视图,用于快速判断哪个 group 最空闲
- status 字段三态:ACTIVE(可读可写) / SYNCING(刚加入,同步中,只读) / OFFLINE(掉线)
---
3. Router:Tracker API
<?php
// src/FDFS/Tracker/Router.php
namespace App\FDFS\Tracker;
use Swoole\Http\Request;
use Swoole\Http\Response;
class Router
{
public function __construct(private Container $c) {}
public function dispatch(Request $req, Response $res): void
{
$res->header('Content-Type', 'application/json');
$uri = $req->server['request_uri'];
try {
$out = match($uri) {
// 客户端 API
'/api/upload/select' => $this->selectForUpload($req),
'/api/download/select' => $this->selectForDownload($req),
'/api/delete/select' => $this->selectAllForDelete($req),
// Storage 节点 API
'/api/storage/register'=> $this->register($req),
'/api/storage/heartbeat'=> $this->heartbeat($req),
'/api/storage/peers' => $this->getPeers($req),
// 管理 API
'/admin/nodes' => $this->c->storages,
'/admin/groups' => $this->c->groupStats,
default => ['code'=>404],
};
// Swoole\Table 直接序列化
if ($out instanceof \Swoole\Table) {
$list = []; foreach ($out as $k=>$v) $list[$k] = $v;
$out = $list;
}
$res->end(json_encode($out));
} catch (\Throwable $e) {
$res->status(500);
$res->end(json_encode(['code'=>500,'msg'=>$e->getMessage()]));
}
}
// 选一个 Storage 来上传:返回组里"主节点"(leader),客户端只往主上传
public function selectForUpload(Request $req): array
{
$size = (int)($req->get['size'] ?? 0);
$hint = $req->get['group'] ?? ''; // 客户端可指定组,否则自动选
// 1. 挑组
$group = $hint;
if ($group === '') {
$group = $this->pickGroup($size);
if (!$group) return ['code'=>503,'msg'=>'no available group'];
}
// 2. 组内挑节点(主节点)
$node = $this->pickStorageInGroup($group, $size);
if (!$node) return ['code'=>503,'msg'=>"no available storage in $group"];
// 3. 同时返回组内副本节点(客户端可选直传,或交给 storage 异步同步)
$peers = $this->peersInGroup($group, $node['node_id']);
return [
'code' => 0,
'group' => $group,
'primary' => [
'node_id' => $node['node_id'],
'host' => $node['host'],
'port' => $node['port'],
'url' => "http://{$node['host']}:{$node['port']}",
],
'peers' => $peers,
// token: storage 收到后校验,防止绕过 Tracker 直接打 storage
'token' => $this->signToken($group, $node['node_id'], $size),
];
}
// 下载:从 file_id 解析 group →选任意 ACTIVE 节点
public function selectForDownload(Request $req): array
{
$fileId = $req->get['file_id'] ?? '';
if (!preg_match('#^(group\d+)/#', $fileId, $m)) {
return ['code'=>400,'msg'=>'invalid file_id'];
}
$group = $m[1];
// 优先选近的(同机房)—简化:随机健康节点
$candidates = [];
foreach ($this->c->storages as $n) {
if ($n['group'] === $group && $n['status'] === 'ACTIVE') $candidates[] = $n;
}
if (!$candidates) return ['code'=>503,'msg'=>"no node in $group"];
$pick = $candidates[array_rand($candidates)];
return [
'code'=>0,
'url'=>"http://{$pick['host']}:{$pick['port']}/$fileId",
];
}
// 删除:返回组内全部活跃节点,客户端依次 DELETE
public function selectAllForDelete(Request $req): array
{
$fileId = $req->get['file_id'] ?? '';
if (!preg_match('#^(group\d+)/#', $fileId, $m)) {
return ['code'=>400];
}
$nodes = [];
foreach ($this->c->storages as $n) {
if ($n['group'] === $m[1] && $n['status'] === 'ACTIVE') $nodes[] = $n;
}
return ['code'=>0,'nodes'=>$nodes,'token'=>$this->signToken($m[1],'',0)];
}
private function pickGroup(int $size): ?string
{
$minMb = $this->c->config['min_free_mb'] + (int)($size / 1024 / 1024);
$best = null; $bestFree = -1;
foreach ($this->c->groupStats as $g => $s) {
if ($s['active_nodes'] < 1) continue;
if ($s['total_free_mb'] < $minMb) continue;
if ($s['total_free_mb'] > $bestFree) {
$best = $g; $bestFree = $s['total_free_mb'];
}
}
return $best;
}
private function pickStorageInGroup(string $group, int $size): ?array
{
$minMb = $this->c->config['min_free_mb'] + (int)($size / 1024 / 1024);
$candidates = [];
foreach ($this->c->storages as $n) {
if ($n['group'] !== $group) continue;
if ($n['status'] !== 'ACTIVE') continue;
if ($n['free_mb'] < $minMb) continue;
$candidates[] = $n;
}
if (!$candidates) return null;
return match($this->c->config['select_policy']) {
'round_robin' => $this->rr($candidates, $group),
'least_used' => $this->leastUsed($candidates),
'weighted' => $this->weighted($candidates),
default => $candidates[array_rand($candidates)],
};
}
private static array $rrCursors = [];
private function rr(array $candidates, string $group): array
{
$i = (self::$rrCursors[$group] ?? -1) + 1;
self::$rrCursors[$group] = $i;
return $candidates[$i % count($candidates)];
}
private function leastUsed(array $c): array
{
usort($c, fn($a,$b) => $b['free_mb'] <=> $a['free_mb']);
return $c[0];
}
private function weighted(array $c): array
{
$sum = 0; foreach ($c as $n) $sum += max(1, $n['weight']);
$pick = random_int(1, $sum); $acc = 0;
foreach ($c as $n) { $acc += max(1, $n['weight']); if ($pick <= $acc) return $n; }
return $c[0];
}
private function peersInGroup(string $group, string $excludeId): array
{
$peers = [];
foreach ($this->c->storages as $n) {
if ($n['group'] === $group && $n['node_id'] !== $excludeId && $n['status'] === 'ACTIVE') {
$peers[] = ['node_id'=>$n['node_id'],'host'=>$n['host'],'port'=>$n['port'],
'url'=>"http://{$n['host']}:{$n['port']}"];
}
}
return $peers;
}
private function signToken(string $group, string $nodeId, int $size): string
{
$exp = time() + 60;
$payload = "$group|$nodeId|$size|$exp";
return base64_encode($payload . '|' . hash_hmac('sha256', $payload, 'tracker-secret-key'));
}
// ===== Storage 节点 API =====
public function register(Request $req): array
{
$d = json_decode($req->rawContent(), true) ?? [];
$nodeId = $d['node_id'] ?? '';
if (!$nodeId) return ['code'=>400,'msg'=>'node_id required'];
$exists = $this->c->storages->exist($nodeId);
$this->c->storages->set($nodeId, [
'node_id' => $nodeId,
'group' => $d['group'] ?? 'group1',
'host' => $d['host'] ?? '',
'port' => (int)($d['port'] ?? 0),
'status' => $exists ? 'ACTIVE' : 'SYNCING', // 新节点先 SYNCING
'total_mb'=> (int)($d['total_mb'] ?? 0),
'free_mb' => (int)($d['free_mb'] ?? 0),
'upload_qps' => 0,
'last_hb' => time(),
'weight' => (int)($d['weight'] ?? 1),
]);
$this->c->rebuildGroupStats();
// 持久化
$this->c->withMySQL(function($db) use ($d, $nodeId) {
$stmt = $db->prepare(
"REPLACE INTO storages(node_id,group_name,host,port,total_mb,weight,enabled,updated_at)
VALUES(?,?,?,?,?,?,1,NOW())"
);
$stmt->execute([$nodeId,$d['group']??'group1',$d['host']??'',
(int)$d['port'],(int)$d['total_mb'],(int)($d['weight']??1)]);
});
return ['code'=>0,'msg'=>'registered'];
}
public function heartbeat(Request $req): array
{
$d = json_decode($req->rawContent(), true) ?? [];
$nodeId = $d['node_id'] ?? '';
$node = $this->c->storages->get($nodeId);
if (!$node) return ['code'=>404,'msg'=>'node not registered'];
$node['free_mb'] = (int)($d['free_mb'] ?? $node['free_mb']);
$node['upload_qps']= (int)($d['upload_qps'] ?? 0);
$node['status'] = $d['status'] ?? 'ACTIVE';
$node['last_hb'] = time();
$this->c->storages->set($nodeId, $node);
return ['code'=>0];
}
public function getPeers(Request $req): array
{
$group = $req->get['group'] ?? '';
$exclude = $req->get['exclude'] ?? '';
return ['code'=>0,'peers'=>$this->peersInGroup($group, $exclude)];
}
}
解释:
- pickGroup →pickStorageInGroup 两级调度:先选最空闲组,再选组内最优节点
- 三种负载均衡策略可热切换:round_robin / least_used / weighted
- signToken:Storage 收到上传请求会验签,防止恶意客户端绕过 Tracker 直接打 Storage
- 新节点 SYNCING 状态:期间只接收同步数据,不参与负载均衡,等同步完才转 ACTIVE
---
4. HealthChecker:心跳超时检测
<?php
// src/FDFS/Tracker/HealthChecker.php
namespace App\FDFS\Tracker;
class HealthChecker
{
public function __construct(private Container $c) {}
public function run(): void
{
while (true) {
\Swoole\Coroutine::sleep(3);
$now = time();
$timeout = $this->c->config['hb_timeout'];
$changed = false;
foreach ($this->c->storages as $id => $n) {
if ($n['status'] === 'OFFLINE') continue;
if ($now - $n['last_hb'] > $timeout) {
$n['status'] = 'OFFLINE';
$this->c->storages->set($id, $n);
error_log("[health] $id timeout, marked OFFLINE");
$changed = true;
// 告警(略)
}
}
if ($changed) $this->c->rebuildGroupStats();
}
}
}
---
B. Storage 节点(实际存数据)
1. Storage Server
<?php
// storage.php —每个 Storage 节点跑一个
declare(strict_types=1);
require __DIR__ . '/vendor/autoload.php';
use Swoole\Http\Server;
use Swoole\Http\Request;
use Swoole\Http\Response;
use App\FDFS\Storage\StorageContext;
use App\FDFS\Storage\Router;
\Swoole\Runtime::enableCoroutine(SWOOLE_HOOK_ALL);
// 节点配置(每个 Storage 部署时改这里)
$ctx = StorageContext::boot([
'node_id' => getenv('NODE_ID') ?: 'storage-1',
'group' => getenv('GROUP') ?: 'group1',
'host' => getenv('HOST') ?: '10.0.0.11',
'port' => (int)(getenv('PORT') ?: 23000),
'tracker' => getenv('TRACKER') ?: 'http://10.0.0.1:22122',
'data_dir' => getenv('DATA_DIR') ?: '/data/fdfs/M00',
'weight' => 1,
]);
$server = new Server('0.0.0.0', $ctx->cfg['port'], SWOOLE_PROCESS);
$server->set([
'worker_num' => swoole_cpu_num() * 2,
'task_worker_num' => 8,
'task_enable_coroutine'=> true,
'enable_coroutine' => true,
'max_request' => 50000,
'hook_flags' => SWOOLE_HOOK_ALL,
'package_max_length'=> 2 * 1024 * 1024 * 1024, // 2GB
'socket_buffer_size'=> 128 * 1024 * 1024,
'buffer_output_size'=> 128 * 1024 * 1024,
'http_parse_post' => false,
'send_yield' => true,
]);
$server->on('WorkerStart', function($s, $wid) use ($ctx) {
$ctx->server = $s;
if ($wid === 0) {
\Swoole\Coroutine::create(fn() => $ctx->heartbeatLoop());
\Swoole\Coroutine::create(fn() => $ctx->retrySyncLoop());
}
});
$server->on('Request', function(Request $req, Response $res) use ($ctx) {
(new Router($ctx))->dispatch($req, $res);
});
// Task:异步副本同步
$server->on('Task', function($s, $task) use ($ctx) {
$ctx->syncToPeers($task->data);
});
$server->on('Finish', fn() => null);
$server->start();
---
2. StorageContext + 心跳 + 副本同步
<?php
// src/FDFS/Storage/StorageContext.php
namespace App\FDFS\Storage;
use Swoole\Coroutine\Http\Client;
use Swoole\Coroutine\Channel;
use Swoole\Http\Server;
class StorageContext
{
public array $cfg;
public ?Server $server = null;
public Channel $syncRetry; // 同步失败重试队列
public int $uploadCount = 0;
public static function boot(array $cfg): self
{
$c = new self();
$c->cfg = $cfg;
if (!is_dir($cfg['data_dir'])) mkdir($cfg['data_dir'], 0755, true);
$c->syncRetry = new Channel(10000);
// 注册到 Tracker
\Swoole\Coroutine\run(function() use ($c) { $c->register(); });
return $c;
}
public function register(): void
{
[$host, $port] = $this->parseTracker();
$client = new Client($host, $port);
$client->setHeaders(['Content-Type'=>'application/json']);
$client->post('/api/storage/register', json_encode([
'node_id' => $this->cfg['node_id'],
'group' => $this->cfg['group'],
'host' => $this->cfg['host'],
'port' => $this->cfg['port'],
'total_mb'=> (int)(disk_total_space($this->cfg['data_dir'])/1024/1024),
'weight' => $this->cfg['weight'],
]));
$client->close();
}
public function heartbeatLoop(): void
{
while (true) {
\Swoole\Coroutine::sleep(5);
try {
[$host, $port] = $this->parseTracker();
$client = new Client($host, $port);
$client->setHeaders(['Content-Type'=>'application/json']);
$client->set(['timeout'=>2]);
$client->post('/api/storage/heartbeat', json_encode([
'node_id' => $this->cfg['node_id'],
'free_mb' => (int)(disk_free_space($this->cfg['data_dir'])/1024/1024),
'upload_qps'=> $this->uploadCount,
'status' => 'ACTIVE',
]));
$client->close();
$this->uploadCount = 0;
} catch (\Throwable $e) {
error_log("[hb] ".$e->getMessage());
}
}
}
public function syncToPeers(array $info): void
{
// info: ['file_id'=>..., 'local_path'=>..., 'peers'=>[...]]
foreach ($info['peers'] as $peer) {
try {
$url = parse_url($peer['url']);
$client = new Client($url['host'], $url['port'], false);
$client->set(['timeout'=>30]);
$client->setHeaders([
'X-Sync-Token' => $this->signSync($info['file_id']),
'X-File-Id' => $info['file_id'],
'Content-Type' => 'application/octet-stream',
]);
$client->setData(file_get_contents($info['local_path'])); // 小文件
$client->post('/replicate');
if ($client->statusCode !== 200) throw new \RuntimeException("sync fail to {$peer['node_id']}");
$client->close();
} catch (\Throwable $e) {
error_log("[sync] retry: ".$e->getMessage());
$this->syncRetry->push(['info'=>$info,'peer'=>$peer,'attempts'=>1], 0.1);
}
}
}
public function retrySyncLoop(): void
{
while (true) {
$item = $this->syncRetry->pop();
if (!$item) continue;
\Swoole\Coroutine::sleep(min(60, 2 ** $item['attempts'])); // 指数退避
try {
$this->syncToPeers(['file_id'=>$item['info']['file_id'],
'local_path'=>$item['info']['local_path'],'peers'=>[$item['peer']]]);
} catch (\Throwable $e) {
if ($item['attempts'] < 8) {
$item['attempts']++;
$this->syncRetry->push($item, 0.1);
} else {
error_log("[sync] giving up: {$item['info']['file_id']} →{$item['peer']['node_id']}");
// 死信(略)
}
}
}
}
public function signSync(string $fileId): string
{
return hash_hmac('sha256', $fileId.'|'.$this->cfg['group'], 'tracker-secret-key');
}
public function verifyTrackerToken(string $token, string $group, string $nodeId, int $size): bool
{
$raw = base64_decode($token);
$parts = explode('|', $raw);
if (count($parts) !== 5) return false;
[$g, $nid, $sz, $exp, $sig] = $parts;
if ($g !== $group) return false;
if ($nid !== '' && $nid !== $nodeId) return false;
if ((int)$exp < time()) return false;
$expectedSig = hash_hmac('sha256', "$g|$nid|$sz|$exp", 'tracker-secret-key');
return hash_equals($expectedSig, $sig);
}
private function parseTracker(): array
{
$u = parse_url($this->cfg['tracker']);
return [$u['host'], $u['port'] ?? 80];
}
}
解释:
- 心跳每 5 秒一次,上报 free_mb 和 upload_qps,Tracker 据此动态调度
- 副本同步异步进行:写完主节点立刻返回客户端,Task 进程慢慢推到副本(写延迟 = 单机延迟)
- 失败指数退避重试:1→2→4→8→16…次失败入死信
- HMAC token 防绕过:Storage 拒绝没有 Tracker 签名的上传
---
3. Storage Router(实际收发文件)
<?php
// src/FDFS/Storage/Router.php
namespace App\FDFS\Storage;
use Swoole\Http\Request;
use Swoole\Http\Response;
class Router
{
public function __construct(private StorageContext $ctx) {}
public function dispatch(Request $req, Response $res): void
{
$uri = $req->server['request_uri'];
$method = $req->server['request_method'];
// 直接下载:/groupX/M00/00/AB/abc.jpg
if ($method === 'GET' && preg_match('#^/group\d+/#', $uri)) {
return $this->serveFile($req, $res, $uri);
}
if ($method === 'HEAD' && preg_match('#^/group\d+/#', $uri)) {
return $this->headFile($res, $uri);
}
// 上传
if ($uri === '/upload' && $method === 'POST') {
return $this->upload($req, $res);
}
// 副本同步接收
if ($uri === '/replicate' && $method === 'POST') {
return $this->replicate($req, $res);
}
// 删除
if ($method === 'DELETE' && preg_match('#^/group\d+/#', $uri)) {
return $this->delete($req, $res, $uri);
}
$res->status(404); $res->end('not found');
}
public function upload(Request $req, Response $res): void
{
// 1. 验证 Tracker 签发的 token
$token = $req->header['x-tracker-token'] ?? '';
$size = strlen($req->rawContent());
if (!$this->ctx->verifyTrackerToken($token, $this->ctx->cfg['group'], $this->ctx->cfg['node_id'], 0)) {
$res->status(403); $res->end('bad token'); return;
}
$ext = pathinfo($req->header['x-file-name'] ?? 'bin', PATHINFO_EXTENSION);
$ext = preg_match('/^[a-z0-9]{1,8}$/i', $ext) ? $ext : 'bin';
// 2. 生成 file_id (FastDFS 风格)
[$fileId, $localPath] = $this->genFileId($ext);
// 3. 落盘
$dir = dirname($localPath);
if (!is_dir($dir)) mkdir($dir, 0755, true);
file_put_contents($localPath, $req->rawContent());
$this->ctx->uploadCount++;
// 4. 异步推副本(从 Tracker 取 peers)
$peers = $this->fetchPeers();
if ($peers) {
$this->ctx->server->task([
'file_id' => $fileId,
'local_path' => $localPath,
'peers' => $peers,
]);
}
// 5. 写元数据(供 HEAD 加速)
@file_put_contents($localPath.'.meta', json_encode([
'size'=>$size,'mime'=>$req->header['content-type'] ?? 'application/octet-stream',
'mtime'=>time(),'etag'=>md5_file($localPath),
]));
$res->end(json_encode(['code'=>0,'file_id'=>$fileId,'size'=>$size]));
}
public function serveFile(Request $req, Response $res, string $uri): void
{
$localPath = $this->ctx->cfg['data_dir'] . $this->stripGroupPrefix($uri);
if (!is_file($localPath)) { $res->status(404); $res->end('not found'); return; }
$size = filesize($localPath);
$start = 0; $end = $size - 1;
$range = $req->header['range'] ?? '';
if (preg_match('/bytes=(\d+)-(\d*)/', $range, $m)) {
$start = (int)$m[1];
$end = $m[2] === '' ? $size - 1 : (int)$m[2];
$res->status(206);
$res->header('Content-Range', "bytes $start-$end/$size");
}
// 读元数据快速回 mime
$meta = is_file($localPath.'.meta') ? json_decode(file_get_contents($localPath.'.meta'), true) : [];
$res->header('Content-Type', $meta['mime'] ?? 'application/octet-stream');
$res->header('Content-Length', (string)($end - $start + 1));
$res->header('Accept-Ranges', 'bytes');
$res->header('ETag', '"'.($meta['etag'] ?? '').'"');
// ★sendfile 零拷贝
$res->sendfile($localPath, $start, $end - $start + 1);
}
public function headFile(Response $res, string $uri): void
{
$localPath = $this->ctx->cfg['data_dir'] . $this->stripGroupPrefix($uri);
if (!is_file($localPath)) { $res->status(404); $res->end(''); return; }
$meta = is_file($localPath.'.meta') ? json_decode(file_get_contents($localPath.'.meta'), true) : [];
$res->header('Content-Length', (string)($meta['size'] ?? filesize($localPath)));
$res->header('Content-Type', $meta['mime'] ?? 'application/octet-stream');
$res->header('ETag', '"'.($meta['etag'] ?? '').'"');
$res->end('');
}
public function replicate(Request $req, Response $res): void
{
// 副本接收:验签 + 落盘到完全相同的路径
$sig = $req->header['x-sync-token'] ?? '';
$fileId = $req->header['x-file-id'] ?? '';
$expected = hash_hmac('sha256', $fileId.'|'.$this->ctx->cfg['group'], 'tracker-secret-key');
if (!hash_equals($expected, $sig)) {
$res->status(403); $res->end('bad sync token'); return;
}
$local = $this->ctx->cfg['data_dir'] . $this->stripGroupPrefix('/'.$fileId);
$dir = dirname($local);
if (!is_dir($dir)) mkdir($dir, 0755, true);
file_put_contents($local, $req->rawContent());
@file_put_contents($local.'.meta', json_encode([
'size'=>strlen($req->rawContent()),
'mime'=>$req->header['content-type'] ?? 'application/octet-stream',
'mtime'=>time(),'etag'=>md5_file($local),
]));
$res->end('OK');
}
public function delete(Request $req, Response $res, string $uri): void
{
$token = $req->header['x-tracker-token'] ?? '';
if (!$this->ctx->verifyTrackerToken($token, $this->ctx->cfg['group'], '', 0)) {
$res->status(403); $res->end('bad token'); return;
}
$local = $this->ctx->cfg['data_dir'] . $this->stripGroupPrefix($uri);
@unlink($local);
@unlink($local.'.meta');
$res->status(204); $res->end('');
}
// file_id: group1/M00/00/AB/wKgABcdef0123456.jpg
private function genFileId(string $ext): array
{
$ts = pack('N', time());
$rand = random_bytes(8);
$name = rtrim(strtr(base64_encode($ts.$rand), '+/', '-_'), '=');
$name = $name . '.' . $ext;
$h = md5($name, true);
$d1 = sprintf('%02X', ord($h[0]));
$d2 = sprintf('%02X', ord($h[1]));
$relativePath = "M00/$d1/$d2/$name";
$fileId = $this->ctx->cfg['group'] . '/' . $relativePath;
$localPath = $this->ctx->cfg['data_dir'] . '/' . substr($relativePath, 4); // 去掉 M00/
return [$fileId, $localPath];
}
private function stripGroupPrefix(string $uri): string
{
// /group1/M00/00/AB/x.jpg →/00/AB/x.jpg (相对 data_dir)
return preg_replace('#^/group\d+/M\d+#', '', $uri);
}
private function fetchPeers(): array
{
$u = parse_url($this->ctx->cfg['tracker']);
$client = new \Swoole\Coroutine\Http\Client($u['host'], $u['port'] ?? 80);
$client->get("/api/storage/peers?group={$this->ctx->cfg['group']}&exclude={$this->ctx->cfg['node_id']}");
$body = json_decode($client->body, true);
$client->close();
return $body['peers'] ?? [];
}
}
解释:
- genFileId:base64(时间戳+随机) 做主键,md5 取前 2 字节做两级散列目录,避免单目录文件数爆炸
- 上传走 token 校验:任何没有 Tracker 签名的上传都被拒,不能绕过调度
- 副本同步用独立的 HMAC,和上传 token 解耦
- sendfile 零拷贝:大文件下载 PHP 进程不占内存
- 元数据 .meta 旁路文件:HEAD 请求不需要 stat 真实文件,性能翻倍
---
C. Client SDK(应用使用)
<?php
// src/FDFS/Client/FdfsClient.php
namespace App\FDFS\Client;
use Swoole\Coroutine\Http\Client;
class FdfsClient
{
public function __construct(private string $trackerUrl) {}
public function upload(string $localPath, string $group = ''): array
{
$size = filesize($localPath);
$body = file_get_contents($localPath);
// 1. 问 Tracker
$u = parse_url($this->trackerUrl);
$t = new Client($u['host'], $u['port']);
$t->get("/api/upload/select?size=$size".($group ? "&group=$group" : ''));
$sel = json_decode($t->body, true);
$t->close();
if (($sel['code'] ?? -1) !== 0) throw new \RuntimeException("tracker: ".$sel['msg']);
// 2. 直传 Storage(不再经 Tracker)
$p = parse_url($sel['primary']['url']);
$s = new Client($p['host'], $p['port']);
$s->setHeaders([
'X-Tracker-Token' => $sel['token'],
'X-File-Name' => basename($localPath),
'Content-Type' => mime_content_type($localPath) ?: 'application/octet-stream',
]);
$s->setData($body);
$s->post('/upload');
$resp = json_decode($s->body, true);
$s->close();
if (($resp['code'] ?? -1) !== 0) throw new \RuntimeException('upload fail');
return [
'file_id' => $resp['file_id'],
'size' => $resp['size'],
'url' => $sel['primary']['url'] . '/' . $resp['file_id'],
];
}
public function download(string $fileId, string $savePath): int
{
$u = parse_url($this->trackerUrl);
$t = new Client($u['host'], $u['port']);
$t->get("/api/download/select?file_id=".urlencode($fileId));
$sel = json_decode($t->body, true);
$t->close();
if (($sel['code'] ?? -1) !== 0) throw new \RuntimeException($sel['msg']);
$p = parse_url($sel['url']);
$s = new Client($p['host'], $p['port']);
$s->get($p['path']);
file_put_contents($savePath, $s->body);
$size = strlen($s->body);
$s->close();
return $size;
}
public function delete(string $fileId): bool
{
$u = parse_url($this->trackerUrl);
$t = new Client($u['host'], $u['port']);
$t->get("/api/delete/select?file_id=".urlencode($fileId));
$sel = json_decode($t->body, true);
$t->close();
foreach ($sel['nodes'] as $n) {
$s = new Client($n['host'], $n['port']);
$s->setHeaders(['X-Tracker-Token' => $sel['token']]);
$s->execute("/$fileId"); // DELETE
// 简化:用低层 HTTP DELETE
$s->setMethod('DELETE');
$s->execute("/$fileId");
$s->close();
}
return true;
}
}
// 使用
\Swoole\Coroutine\run(function() {
$c = new FdfsClient('http://127.0.0.1:22122');
$info = $c->upload('/tmp/photo.jpg');
echo $info['file_id'] . "\n"; // group1/M00/00/AB/wKg...jpg
$c->download($info['file_id'], '/tmp/downloaded.jpg');
});
---
D. DDL
CREATE TABLE storages (
node_id VARCHAR(32) PRIMARY KEY,
group_name VARCHAR(32) NOT NULL,
host VARCHAR(64) NOT NULL,
port INT NOT NULL,
total_mb BIGINT,
weight INT DEFAULT 1,
enabled TINYINT DEFAULT 1,
updated_at DATETIME,
KEY idx_group(group_name)
);
CREATE TABLE files (
file_id VARCHAR(128) PRIMARY KEY,
group_name VARCHAR(32),
size BIGINT,
mime VARCHAR(64),
uploaded_by VARCHAR(64),
created_at DATETIME,
KEY idx_group(group_name)
);
---
五、性能参考(单机 4C8G)
┌──────────────┬────────────────────┬─────────────────┐
│ 角色 │ 指标 │ 数值 │
├──────────────┼────────────────────┼─────────────────┤
│ Tracker │ 调度 QPS │ 3-5 万 │
├──────────────┼────────────────────┼─────────────────┤
│ Tracker │ P99 延迟 │ < 5ms │
├──────────────┼────────────────────┼─────────────────┤
│ Storage │ 小文件上传 QPS │ 3000-5000 │
├──────────────┼────────────────────┼─────────────────┤
│ Storage │ 下载并发(sendfile) │ 5000+ 长连接 │
├──────────────┼────────────────────┼─────────────────┤
│ Storage │ 大文件吞吐 │ 网卡线速 │
├──────────────┼────────────────────┼─────────────────┤
│ 副本同步延迟 │ │ < 100ms(同机房) │
├──────────────┼────────────────────┼─────────────────┤
│ 心跳→OFFLINE│ │ < 15s │
└──────────────┴────────────────────┴─────────────────┘
---
六、踩坑提示
┌───────────────────────────────────┬────────────────────────────────────────────────────┐
│ 坑 │ 解决 │
├───────────────────────────────────┼────────────────────────────────────────────────────┤
│ 大文件 rawContent() 把进程吃 OOM │ 大文件改用 Storage 直传(避开 Tracker)+ 流式 │
├───────────────────────────────────┼────────────────────────────────────────────────────┤
│ 副本同步成功率不到 100% │ 必须有重试队列 + 死信,人工对账 │
├───────────────────────────────────┼────────────────────────────────────────────────────┤
│ 新节点加入立刻参与负载 →数据缺失 │ 先 SYNCING 状态,等同步完才 ACTIVE │
├───────────────────────────────────┼────────────────────────────────────────────────────┤
│ 单目录文件数过百万 │ 用两级散列目录(本方案已有) │
├───────────────────────────────────┼────────────────────────────────────────────────────┤
│ 同步把网络打满 │ Storage 加同步限速(令牌桶) │
├───────────────────────────────────┼────────────────────────────────────────────────────┤
│ Tracker 单点 │ Tracker 多副本 + 客户端轮询多个地址 │
├───────────────────────────────────┼────────────────────────────────────────────────────┤
│ Storage 磁盘满 →文件半写入 │ 写之前 disk_free_space 预判 │
├───────────────────────────────────┼────────────────────────────────────────────────────┤
│ file_id 冲突 │ 时间戳 + 8 字节随机,冲突概率 1e-18;再不放心加 stat │
├───────────────────────────────────┼────────────────────────────────────────────────────┤
│ 元数据丢失 │ .meta 文件 + DB files 表双写 │
├───────────────────────────────────┼────────────────────────────────────────────────────┤
│ 同机房优先(降低跨机房带宽) │ 心跳上报 region 字段,调度时按 client IP 匹配 │
└───────────────────────────────────┴────────────────────────────────────────────────────┘
---
七、安全清单
┌───────────────────────────────────┬───────────────────────────────────┐
│ 风险 │ 防御 │
├───────────────────────────────────┼───────────────────────────────────┤
│ 客户端绕过 Tracker 直接打 Storage │ Tracker token 必校验(本方案 HMAC) │
├───────────────────────────────────┼───────────────────────────────────┤
│ 文件 ID 被猜测枚举 │ 8 字节随机,熵 ≥64 bit │
├───────────────────────────────────┼───────────────────────────────────┤
│ 上传任意路径(目录穿越) │ stripGroupPrefix 严格正则,拒绝 .. │
├───────────────────────────────────┼───────────────────────────────────┤
│ 跨节点同步被伪造 │ 独立 HMAC sync-token │
├───────────────────────────────────┼───────────────────────────────────┤
│ 大文件吃满磁盘 │ 单节点限额 + 总配额 + 告警 │
├───────────────────────────────────┼───────────────────────────────────┤
│ 公网直接访问 Storage │ Storage 内网部署,通过 CDN / 网关 │
├───────────────────────────────────┼───────────────────────────────────┤
│ Tracker 接口被刷 │ IP 限流 + 业务侧 token │
└───────────────────────────────────┴───────────────────────────────────┘
---
八、可扩展方向
1. EC 纠删码:3 数据 + 2 校验,1.67 倍空间换 N+2 容灾(类 Ceph)
2. 冷热分层:30 天未访问的文件搬到便宜的对象存储(OSS)
3. 客户端断点续传:Storage 支持 PUT Range,客户端分块续传
4. CDN 集成:/download/select 返回 CDN URL,Storage 做回源
5. 去重(content-addressed):file_id = sha256(content),相同内容只存一份
6. 多机房复制:跨机房副本 + 异步追平
7. 图片处理:URL 带 ?resize=100x100,Storage 实时缩放
8. 生命周期:?expire=86400 上传时打标,定时清理
9. 审计 + DLP:上传内容扫描敏感词
10. WebDAV / S3 协议对外:加协议适配器,让任何客户端都能用
---
九、和 FastDFS / SeaweedFS 的真实差距
┌────────────┬────────────────┬───────────────────┬───────────────────────────────┐
│ 维度 │ FastDFS (C) │ SeaweedFS (Go) │ 本方案 (PHP+Swoole) │
├────────────┼────────────────┼───────────────────┼───────────────────────────────┤
│ 性能(单机) │ 极高 │ 极高 │ 中(够中小规模) │
├────────────┼────────────────┼───────────────────┼───────────────────────────────┤
│ 协议 │ 私有二进制 │ HTTP/gRPC │ HTTP │
├────────────┼────────────────┼───────────────────┼───────────────────────────────┤
│ 部署复杂度 │ 中 │ 低 │ 低 │
├────────────┼────────────────┼───────────────────┼───────────────────────────────┤
│ 扩展灵活性 │ 改 C 难 │ 改 Go 中 │ 改 PHP 快(本方案强项) │
├────────────┼────────────────┼───────────────────┼───────────────────────────────┤
│ 业务集成 │ 适配麻烦 │ 中 │ 天然契合 PHP 业务(本方案强项) │
├────────────┼────────────────┼───────────────────┼───────────────────────────────┤
│ 适合场景 │ 大规模通用文件 │ 大规模 + 现代特性 │ 中小团队 + 业务深度定制 │
└────────────┴────────────────┴───────────────────┴───────────────────────────────┘
更多推荐
0
0- 0
已为社区贡献83条内容
所有评论(0)