PHP实现Workerman实例 高性能PHP Socket即时通讯框架

composer require topthink/think-worker

php think worker:server

return [
    'socket'    =>  'http://127.0.0.1:8000',
    'name'      =>  'thinkphp',
    'count'     =>  4,
    'onMessage' =>  function($connection, $data) {
        $connection->send(json_encode($data));
    },
];

return [
    'worker_class'  =>  'app\http\Workerman',
];

<?php
namespace app\http;
use think\Log;
use think\worker\Server;
use Workerman\Lib\Timer;
class Worker extends Server
{
    // socket 端口
    protected $worker;
    protected $processes = 1; // 一个进程数
    protected $socket = '';
    protected $uidConnections = []; // 用户连接
    protected $heartbeat_time = 55; // 心跳间隔检测
    protected $send_heartbeat_time = 10; // 发送心跳间隔
    protected $time_countdown = 0; // 计算时间
    public function __construct()
    {
        $this->socket = 'http://0.0.0.0:2345';
        parent::__construct();
    }
    /**
     * 收到信息
     */
    public function onMessage($connection, $data)
    {
        $connection->lastMessageTime = time();
        $connection->send(json_encode($data));
    }
    /**
     * 当连接建立时触发的回调函数
     */
    public function onConnect($connection)
    {
    }
    /**
     * 当连接断开时触发的回调函数
     */
    public function onClose($connection)
    {
        // 删除
        if (isset($connection->uid) && isset($this->uidConnections[$connection->uid])) {
            //unset($this->uidConnections[$connection->uid][$connection->id]);
            unset($this->uidConnections[$connection->uid]);
        }
    }
    /**
     * 当客户端的连接上发生错误时触发
     */
    public function onError($connection, $code, $msg)
    {
        ////echo "error $code $msg\n";
    }
    /**
     * 发送给所有人
     */
    private function sendAll($data, $flg = false, $type = '')
    {
        $msg = json_encode([
            'event' => '#publish',
            'data' => $data,
            'type' => $type
        ]);
        $connection_check = [];
        // 发送给登录的用户
        // foreach ($this->uidConnections as $connection) {
        //     $end = current($connection);
        //     $end->send($msg);
        //     $connection_check[] = $end->id;
        // }
        // 发送给所有人
        foreach ($this->worker->connections as $connection) {
            if (in_array($connection->id, $connection_check)) {
                continue;
            }
            $connection->send($msg);
        }
    }
    /**
     * 发送给指定用户
     */
    private function sendMsg($uid, $data, $type)
    {
        $msg = json_encode([
            'type' => $type,  //待处理todo 多种type消息
            'data' => $data,
            'alert' => true,
        ]);
        $uid = intval($uid);
        // 判断是否有在线的连接
//        if (isset($this->uidConnections[$uid])) {
//            $end = $this->uidConnections[$uid];
//            $end->send($msg);
//        }
        // 判断是否有在线的连接
        if (isset($this->uidConnections[$uid]) && $this->uidConnections[$uid]) {
            //$end = end($this->uidConnections[$uid]);
            //$end->send($msg);
            foreach ($this->uidConnections[$uid] as $connection) {
                $connection->send($msg);
            }
        }
    }
    /**
     * 每个进程启动
     * @param $worker
     */
    public function onWorkerStart($worker)
    {
        //每秒执行
        Timer::add(1, function () use ($worker) {
            $this->time_countdown++;
            $time_now = time();
            foreach ($worker->connections as $connection) {
                $connection->send('time_countdown:' . $this->time_countdown);
                // 有可能该connection还没收到过消息，则lastMessageTime设置为当前时间
                // 统一下发心跳
                if ($this->time_countdown % $this->send_heartbeat_time == 0) {
                    $connection->send(json_encode([
                        'type' => 'hello time',
                    ]));
                }
                if (empty($connection->lastMessageTime)) {
                    $connection->lastMessageTime = $time_now;
                    continue;
                }
                // 上次通讯时间间隔大于心跳间隔，则认为客户端已经下线，关闭连接
                if ($time_now - $connection->lastMessageTime > $this->heartbeat_time) {
                    $connection->close();
                }
            }
        });
    }
}

// $file = request()->file('image');
$file_data  =  $_FILES[0]['file_data'];
//$tmp_file  = tempnam('','tm_');  这种写法最终保存时扩展名为.tmp
$tmp_file  = sys_get_temp_dir().'/'.uniqid().'.'.explode('/',$_FILES[0]['file_type'])[1];
file_put_contents($tmp_file,$file);
$file  =  new  File($tmp_file);
$savename  =  Filesystem::putFile('upload',$file);
echo  $savename;

php think make:command Hello hello

protected function configure()
  {
    // 指令配置
    $this->setName('convert')
      ->addArgument('action', Argument::OPTIONAL, "start|stop|restart|reload|status|connections", 'start')
      ->addOption('mode', 'm', Option::VALUE_OPTIONAL, 'Run the workerman server in daemon mode.')
      ->setDescription('the workerman command');
  }
  protected function execute(Input $input, Output $output)
  {
    // 指令输出
    $output->writeln('convert start');
    $action = $input->getArgument('action');
    $mode = $input->getOption('mode');
    // 重新构造命令行参数,以便兼容workerman的命令
    global $argv;
    $argv = [];
    array_unshift($argv, 'think', $action);
    if ($mode == 'd') {
      $argv[] = '-d';
    } else if ($mode == 'g') {
      $argv[] = '-g';
    }
    // 在这里放心的实例化worker,
    // 就像参照workerman文档写一样,
    // 无非在workerman的文档里,代码是新建纯php文件,但在这里,写到了一个方法里.
    $worker_1 = new Worker();
    $worker_2 = new Worker();
    Worker::runAll();
  }

//临时运行
php think hello start
//后台运行
php think hello start --mode d