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


目录
相关文章
|
1月前
|
数据采集 前端开发 PHP
深入PHP内核:探索高性能异步编程
【2月更文挑战第24天】在现代Web开发中,提供快速响应和高并发处理能力是至关重要的。传统的同步阻塞模型已无法满足现代互联网服务的需求。本文将深入探讨PHP语言中的异步编程概念,介绍其原理、实现方式以及如何通过异步编程提升应用性能。我们将剖析PHP的多线程扩展,如pthreads,以及事件驱动编程模型,重点关注它们在处理大量并发连接时的优势和潜在问题。此外,文中还将展示如何使用Swoole等高级框架来构建高效的异步应用。最后,我们将讨论异步编程在实际应用中的挑战与未来发展趋势。
|
28天前
|
缓存 算法 PHP
深入PHP内核:探索高性能代码优化策略
【5月更文挑战第27天】 在本文中,我们将深入探讨PHP的内核机制,并分析如何通过理解其工作原理来优化我们的代码。文章将聚焦于几个关键领域,包括解释器执行流程、内存管理、以及代码编写最佳实践等。我们的目标是为开发者提供实用的指导,帮助他们写出更加高效和稳定的PHP应用程序。
|
1月前
|
网络协议 Java API
Python网络编程基础(Socket编程)Twisted框架简介
【4月更文挑战第12天】在网络编程的实践中,除了使用基本的Socket API之外,还有许多高级的网络编程库可以帮助我们更高效地构建复杂和健壮的网络应用。这些库通常提供了异步IO、事件驱动、协议实现等高级功能,使得开发者能够专注于业务逻辑的实现,而不用过多关注底层的网络细节。
|
1月前
|
缓存 自然语言处理 监控
深入PHP内核:探索高性能脚本编程的秘密
【4月更文挑战第30天】 在现代Web开发中,PHP作为一种流行的服务器端脚本语言,其性能优化一直是开发者关注的焦点。本文将深入探讨PHP内核架构,分析影响PHP脚本性能的关键因素,并提出一系列提升执行效率的策略。我们将从语言解释器的角度出发,剖析词法分析、语法分析和执行机制,同时考虑内存管理和代码优化的实践技巧。通过本文的阅读,读者能够对PHP的性能调优有更深层次的理解,并在实际项目中运用这些知识以实现高效的脚本运行。
|
1月前
|
PHP 开发者
深入PHP内核:探索高性能脚本语言的秘诀
【4月更文挑战第28天】 在现代Web开发领域,PHP一直是一个不可或缺的服务器端脚本语言。尽管面临诸多新兴技术的挑战,PHP凭借其独特的灵活性、强大的社区支持和持续的性能优化,依然稳居市场重要地位。本文将深入探讨PHP的核心特性,分析其在性能方面的优化策略,并探讨如何利用这些策略来编写更高效的代码。我们将从PHP的执行机制入手,逐步剖析其内存管理、并发处理以及扩展性等关键技术点,为开发者提供提升应用性能的实用指导。
|
1月前
|
数据可视化 Java 关系型数据库
基于java Swing 框架使用socket技术开发的即时通讯系统【源码+数据库】
基于java Swing 框架使用socket技术开发的即时通讯系统【源码+数据库】
|
7月前
|
Web App开发 存储 前端开发
Golang微服务框架kratos实现Socket.IO服务
Socket.IO 是一个面向实时 web 应用的 实时通讯库。它使得服务器和客户端之间实时双向的通信成为可能。底层使用EngineIO。SocketIO的的客户端使用Engine.IO-Client,服务端使用Engine.IO实现。
59 0
|
12月前
|
JavaScript
socket.io实现即时通讯
socket.io实现即时通讯
72 0
|
Ubuntu TensorFlow 算法框架/工具
基于 socket 的即时通讯文件传输聊天软件
基于 socket 的即时通讯文件传输聊天软件
115 0
基于 socket 的即时通讯文件传输聊天软件
|
Linux
socket简单文件传输实例
今天在linux系统上写了一个简单的socket文件传输的例子,共享出来仅供参考。例子很简单,只要初学者了解简单的几个socket相关函数和文件函数就能写的出来。
116 0
socket简单文件传输实例