基于："workerman/workerman": "^4.0"

文件上传服务器，可以支持并发上传大文件。基于Workerman。

在命令行中运行

php server.php start

向服务器发送文件

php client.php -h 127.0.0.1 -p 2347 -f  file.tar.gz

运行服务器程序

server.php

<?php
require_once __DIR__.'/vendor/autoload.php';

require_once __DIR__.'/classes/WorkermanUploadServer.php';

$svr = new WorkermanUploadServer();
//设置上传文件的存储目录
$svr->setRootPath('/tmp/');
//设置允许上传的文件最大尺寸
$svr->setMaxSize(100000*1024*1024);
$svr->run();

实现类

<?php
use Workerman\Worker;
use Workerman\Connection\TcpConnection;

class WorkermanUploadServer
{
    protected $worker;
    protected $files;
    protected $root_path = '/tmp/';
    protected $override = false;
    protected $max_file_size = 100000000; //100M

    public function setRootPath(string $path)
    {
        if (!is_dir($path)) 
        {
            throw new Exception(__METHOD__.": $path is not exists.");
        }
        $this->root_path = $path;
    }

    public function setMaxSize(int $max_file_size)
    {
        $this->max_file_size = (int)$max_file_size;
        if ($this->max_file_size <= 0)
        {
            throw new Exception(__METHOD__.": max_file_size is zero.");
        }
    }

    public function onConnect(TcpConnection $connection)
    {
        echo "new upload client[{$connection->id}] connected.\n";
    }

    protected function message(TcpConnection $connection, $code, $msg)
    {
        $connection->send(json_encode(array('code' => $code, 'msg' => $msg)));
        echo "[-->$connection->id]\t$code\t$msg\n";

        if ($code != 0) {
            $connection->close();
        }

        return true;
    }
    public function onMessage(TcpConnection $connection, $data)
    {
        $fd = md5($connection->id);
        if (empty($this->files[$fd]))  {
            $req = json_decode($data, true);
            if($req == false) {
                return $this->message($connection, 400, 'Error Request');
            } elseif (empty($req['size']) || empty($req['name'])) {
                return $this->message($connection, 500, 'require file name and size.');
            } elseif ($req['size'] > $this->max_file_size) {
                return $this->message($connection, 501, 'over the max_file_size. ' . $this->max_file_size);
            }

            $file = $this->root_path . '/' . $req['name'];
            $dir = realpath(dirname($file));
            if (!$dir or strncmp($dir, $this->root_path, strlen($this->root_path)) != 0) {
                return $this->message($connection, 502, "file path[$dir] error. Access deny.");
            } elseif ($this->override && is_file($file)) {
                return $this->message($connection, 503, 'file exists. Server not allowed override');
            }

            $fp = fopen($file, 'w+');
            if (!$fp) {
                return $this->message($connection, 504, 'can open file.');
            } else {
                $this->message($connection, 0, 'transmission start');
                $this->files[$fd] = array('fp' => $fp, 'file' => $file, 'size' => $req['size'], 'recv' => 0);
            }

        } else {
            //传输已建立
            $info = & $this->files[$fd];
            $fp = $info['fp'];
            $file = $info['file'];
            if (!fwrite($fp, $data)) {
                $this->message($connection, 600, "fwrite failed. transmission stop.");
                unlink($file);
            } else {
                $info['recv'] += strlen($data);
                if ($info['recv'] >= $info['size']) {
                    $this->message($connection, 0, "Success, transmission finish. Close connection.");
                    unset($this->files[$fd]);
                }
            }
        }

        return true;
    }

    public function onClose(TcpConnection $connection)
    {
        $fd = $connection->id;
        unset($this->files[$fd]);
        echo "connection client[{$connection->id}] closed\n";
    }

    public function run()
    {
        // 运行在主进程
        $tcp_worker = new Worker("tcp://0.0.0.0:2347");

        $tcp_worker->count = 1;

        $tcp_worker->onWorkerStart = function (Worker $worker) {
            echo "upload server running ...\n";
        };

        $tcp_worker->onConnect = array($this, 'onConnect');

        $tcp_worker->onClose = array($this, 'onClose');

        $tcp_worker->onMessage = array($this, 'onMessage');

        $this->worker = $tcp_worker;
        Worker::runAll();
    }
}

client.php

#!/usr/local/bin/php
<?php
$args = getopt("p:h:f:t");
// 检查上传文件路径参数
if (empty($args['h']) or empty($args['f'])) {
    echo "Usage: php {$argv[0]} -h server_ip -p server_port -f file -t timeout\n";
    exit;
}
if (empty($args['p'])){
    $args['p'] = 9507;
}

if (empty($args['t'])) {
    $args['t'] = 30;
}

$file = $args['f'];

// 上传地址
$address = $args['h'].":".$args['p'];
// 上传文件路径
$file_to_transfer = trim($file);
// 上传的文件本地不存在
if (!is_file($file_to_transfer)) {
    exit("Error: file  $file_to_transfer not exist\n");
}

// 建立socket连接
$client = stream_socket_client($address, $errno, $errmsg);
if (!$client) {
    exit("$errmsg\n");
}
stream_set_blocking($client, 1);

$file_size = filesize($file_to_transfer);
// 文件名
$file_name = basename($file_to_transfer);
// 文件二进制数据
// $file_data = file_get_contents($file_to_transfer);
// base64编码
// $file_data = base64_encode($file_data);
// 数据包
// $package_data = array(
//     'file_name' => $file_name,
//     'file_data' => $file_data,
// );

$package_data = array(
    'name' => $file_name,
    'size' => $file_size
);

// 协议包 json+回车
$package = json_encode($package_data)."\n";
// 执行上传
fwrite($client, $package);
getResponse($client);

echo "Start transport. file={$file_to_transfer}, size={$file_size}\n";

$fp = fopen($file_to_transfer, 'r+');
if (!$fp) {
    die("Error: open $file failed.\n");
}

$i = 0;
while(!feof($fp))
{
    $read = fread($fp, 1024);
    if (!fwrite($client, $read)) {
        exit("send failed.\n");
    }
}
getResponse($client);
echo "Success. send_size = $file_size\n";

function getResponse($client){
    // 打印结果
    $recv =fread($client, 8192);
    var_dump($recv);
}