开发者社区> owenzhang> 正文

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

简介: PHP实现Workerman实例 高性能PHP Socket即时通讯框架
+关注继续查看

安装

首先通过 composer 安装

composer require topthink/think-worker

使用

SocketServer

在命令行启动服务端(需要2.0.5+版本)

php think worker:server

默认会在0.0.0.0:2345开启一个websocket服务。

如果需要自定义参数,可以在config/worker_server.php中进行配置,包括:

配置参数

描述

protocol

协议

host

监听地址

port

监听端口

socket

完整的socket地址

并且支持workerman所有的参数(包括全局静态参数)。

也支持使用闭包方式定义相关事件回调。

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

复制

也支持使用自定义类作为Worker服务入口文件类。例如,我们可以创建一个服务类(必须要继承 think\worker\Server),然后设置属性和添加回调方法

然后在worker_server.php中增加配置参数:

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

app\http\Workerman.php

<?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();
                }
            }
        });
    }
}

关于上传文件

当按照默认的worker做http服务器时,并不能直接使用request()->file('image')来获得上传的文件,具体可以参考workerman的上传文件第6点.因此只能迂回的使用Filesystem.无论怎样,不影响其getMime()等方法的正确性.

// $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;

自定义workerman指令

有时候我们希望使用think的命令行运行workerman,这里做一个介绍,

1:先新建一个指令,参考文档:自定义指令,比如新建命令:

php think make:command Hello hello

2:复制下面的代码到指令里,覆盖原始的configureexecute方法

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();
  }

3:运行的时候,使用如下命令:

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


版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
AES-128-CBC-Pkcs7Padding加密PHP实例
AES-128-CBC-Pkcs7Padding加密PHP实例
71 0
PHP实现php-amqplib/php-amqplib实例RabbitMq
PHP实现php-amqplib/php-amqplib实例RabbitMq
39 0
PHP使用topthink/think-queue消息队列实例
PHP使用topthink/think-queue消息队列实例
67 0
PHP实现定时任务hellogerard/jobby实例
PHP实现定时任务hellogerard/jobby实例
39 0
PHP实现elasticsearch/elasticsearchy实例
PHP实现elasticsearch/elasticsearchy实例
30 0
PHP实现TCP实例
PHP实现TCP实例
16 0
php结合redis实现高并发下的抢购、秒杀功能的实例
php结合redis实现高并发下的抢购、秒杀功能的实例
57 0
PHP实现令牌桶限流Redis list列表 Lpush rpop 实现令牌桶 - 限流 PHP实例
PHP实现令牌桶限流Redis list列表 Lpush rpop 实现令牌桶 - 限流 PHP实例
164 0
iframe_demo实例:消息发送(PHP版本)
iframe_demo实例:消息发送(PHP版本)
72 0
基于PHP和MySQL的新闻发布系统——【功能优化】
基于PHP和MySQL的新闻发布系统——【功能优化】
33658 0
+关注
owenzhang
I actually hate programming, but I love solving problems. Phper &amp; Gopher. Email: owen@owenzhang.com
文章
问答
文章排行榜
最热
最新
相关电子书
更多
PHP安全开发_从白帽角度做安全
立即下载
PHP在机器学习上的应用及云深度学习平台的架构设计与实现
立即下载
复杂PHP系统性能瓶颈排查及优化
立即下载