php - 我需要在 PHP React websocket 事件循环(也许是多线程？)的内部实现一个等待计时器

Question

我有一个正在构建游戏的 websocket 应用程序，该应用程序构建在使用 React 事件循环的 Ratchet 上。在这个脚本的开头，我已经想出了如何实现一个周期性定时器，每秒向游戏发送一个脉冲，然后执行滴答和战斗回合。这很好用。

但是，我最近意识到我还需要添加“滞后”客户端或暂停函数执行的功能。例如，如果一个玩家被惊呆了，或者我想让一个 NPC 等待 1.5 秒再回复触发器以获得更“真实”的对话感觉。

这个功能是内置在 React 库中的，还是我必须通过其他方式实现的？经过一些研究，我可能正在寻找 pthreads，请参阅此问题/答案:How can one use multi threading in PHP applications

为了更清楚我想要实现的目标，以这段代码为例:

    function onSay($string)
{
    global $world;

    $trigger_words = array(
        'hi',
        'hello',
        'greetings'
    );

    $triggered = false;

    foreach($trigger_words as $trigger_word)
    {
        if(stristr($string, $trigger_word))
        {
            $triggered = true;
        }
    }

    if($triggered)
    {
        foreach($world->players as $player)
        {
            if($player->pData->in_room === $this->mobile->in_room)
            {
                sleep(1);
                $this->toChar($player, $this->mobile->short . " says '`kOh, hello!``'");
            }
        }
    }
}

显然，这不起作用，因为 sleep(1) 函数将暂停整个服务器进程。

任何见解将不胜感激。谢谢!

更新:我的服务器脚本:

require 'vendor/autoload.php';
require 'src/autoload.php';
use Ratchet\MessageComponentInterface;
use Ratchet\ConnectionInterface;
use Ratchet\Server\IoServer;
use Ratchet\Http\HttpServer;
use Ratchet\WebSocket\WsServer;
use React\Socket\Server as Reactor;
use React\EventLoop\Factory as LoopFactory;;

$world = new WorldInterface();

class Server implements MessageComponentInterface
{   
    public function __construct(React\EventLoop\LoopInterface $loop) 
    {
        $update = new Update();
        $update->doTick();

        $loop->addPeriodicTimer(1, function() 
        {
            $this->doBeat();    
        });
    }

public function onOpen(ConnectionInterface $ch) 
{
    global $world;
    $world->connecting[$ch->resourceId] = $ch;
    $ch->CONN_STATE = "GET_NAME";
    $ch->pData = new stdClass();
    $ch->send("Who dares storm our wayward path? ");
}

public function onMessage(ConnectionInterface $ch, $args) 
{   
    if($ch->CONN_STATE == "CONNECTED")
    {
        $ch->send("> " . $args . "\n");
        $interpreter = new Interpreter($ch);
        $interpreter->interpret($args);
    }
    else
    {
        $ch->send($args);
        $login = new Login($ch, $args);
        $login->start();
    }

}

public function onClose(ConnectionInterface $ch) 
{
    global $world;

    if(isset($ch->pData->name))
    {
        if(isset($world->players[$ch->pData->name]))
        {
            echo "Player {$ch->pData->name} has disconnected\n";
            unset($world->players->{$ch->pData->name});
        }
    }

    if(isset($world->connecting->{$ch->resourceId}))
    {
        echo "Connection " . $ch->resourceId . " has disconnected.";
        unset($world->connecting->{$ch->resourceId});
    }
}

public function onError(ConnectionInterface $conn, \Exception $e) 
{
    echo "An error has occurred: {$e->getMessage()}\n";
    $conn->close();
}

public function doBeat()
{
    global $world;
    ++$world->beats;

    foreach($world->process_queue as $trigger_beat => $process_array)
    {
        // if the beat # that the function should fire on is less than,
        // or equal to the current beat, fire the function.
        if($trigger_beat <= $world->beats)
        {
            foreach($process_array as $process)
            {
                $class = new $process->class();
                call_user_func_array(array($class, $process->function), $process->params);
            }

            // remove it from the queue
            unset($world->process_queue[$trigger_beat]);
        }
        // else, the beat # the function should fire on is greater than the current beat, 
        // so break out of the loop.
        else
        {
            break;
        }
    }

    if($world->beats % 2 === 0)
    {
        $update = new Update();
        $update->doBeat();
    }
}
}

$loop = LoopFactory::create();
$socket = new Reactor($loop);
$socket->listen(9000, 'localhost');
$server = new IoServer(new HttpServer(new WsServer(new Server($loop))),   $socket, $loop);
$server->run();

Answer 1

好吧，所以我假设因为这个问题仍然没有答案，所以没有“简单”的解决方案融入到 React 事件循环中，尽管我很乐意在这一点上是错误的。在那之前，我想我会发布我的解决方案。

注意:我不知道这样做的含义是什么。我不知道它的可扩展性如何。它未经在具有多个进程和播放器的实时环境中进行测试。

不过，我认为这是一个不错的解决方案。我的特定游戏适合 20 - 30 人的玩家群，所以我认为我可能面临的唯一问题是一堆排队的 Action 是否在同一秒触发。

到代码!

我(不久前)做的第一件事是在服务器启动时添加一个周期性计时器:

public function __construct(React\EventLoop\LoopInterface $loop) 
{
    $update = new Update();
    $update->doTick();

    $loop->addPeriodicTimer(1, function() 
    {
        $this->doBeat();    
    });
}

我的“世界”类中也有一些全局变量:

// things in the world
public $beats = 0;
public $next_tick = 45;
public $connecting = array();
public $players = array();
public $mobiles = array();
public $objects = array();
public $mobs_in_rooms = array();
public $mobs_in_areas = array();
public $in_combat = array(
    'mobiles' => array(),
    'players' => array()
);
public $process_queue;

注意 beats 和 process_queue。

我的 doBeat() 函数如下所示:

public function doBeat()
{
    global $world;
    ++$world->beats;

    foreach($world->process_queue as $trigger_beat => $process_array)
    {
        // if the beat # that the function should fire on is less than,
        // or equal to the current beat, fire the function.
        if($trigger_beat <= $world->beats)
        {
            foreach($process_array as $process)
            {
                $class = new $process->class();
                call_user_func_array(array($class, $process->function), $process->params);
            }

            // remove it from the queue
            unset($world->process_queue[$trigger_beat]);
        }
        // else, the beat # the function should fire on is greater than the current beat, 
        // so break out of the loop.
        else
        {
            break;
        }
    }

    print_r(array_keys($world->process_queue));

    if($world->beats % 2 === 0)
    {
        $update = new Update();
        $update->doBeat();
    }
}

现在，在我的全局“World”对象上，我有几个其他函数:

function addToProcessQueue($process_obj)
{
    //adds the process object to an array of the beat #
    //when it should be triggered on process_queue.

    $this->process_queue[(int)$process_obj->trigger_beat][] = $process_obj;
    ksort($this->process_queue);
}

function createProcessObject($array)
{
    $process_obj = new stdClass();

    if(isset($array['function']))
    {
        $process_obj->function = $array['function'];
    }
    else
    {
        echo "All process requests must define a function to call defined as a key named 'function' on the array you pass.";
    }

    if(isset($array['class']))
    {
        $process_obj->class = $array['class'];
    }
    else
    {
        echo "All process requests must define a class to call defined as a key named 'class' on the array you pass.";
    }

    if(isset($array['params']))
    {
        $process_obj->params = $array['params'];
    }
    else
    {
        $process_obj->params = array();
    }

    if(isset($array['char']))
    {
        $process_obj->char = $array['char'];
    }
    else
    {
        $process_obj->char = false;
    }

    if(isset($array['trigger_beat']) && is_numeric($array['trigger_beat']))
    {
        $process_obj->trigger_beat = $array['trigger_beat'];
    }
    else
    {
        echo "All process requests must define a trigger_beat. \n"
        . "Use world->beats to get current beat and add your wait time onto it. \n"
                . "Trigger beat MUST be an integer. \n";
    }

    $this->addToProcessQueue($process_obj);
}

现在要向队列中添加一个进程，这是我的新移动“onSay()”命令:

function onSay($string)
{
    global $world;

    $trigger_words = array(
        'hi',
        'hello',
        'greetings'
    );

    $triggered = false;

    foreach($trigger_words as $trigger_word)
    {
        if(stristr($string, $trigger_word))
        {
            $triggered = true;
        }
    }

    if($triggered)
    {
        $process_array = array(
            'trigger_beat' => $world->beats + 2,
            'function' => 'toRoom',
            'class' => 'PlayerInterface',
            'params' => array($this->mobile->in_room, $this->mobile->short . " says '`kOh, hello!``'")
        );

        $world->createProcessObject($process_array);
    }
}

因此，如果手机听到“hi”、“hello”或“greetings”，“toRoom”函数(向同一房间中的每个角色发送一个字符串)将被添加到进程队列中并触发 2从执行原始函数开始的秒数。

我希望所有这些都有意义，如果有人知道在 php 和事件循环中完成此类事情的更好方法，请回答/评论。我并没有像我上面所说的那样将其标记为“正确”，我不知道它在生产中的效率如何。