用PHP实现高并发服务器

phpize
./configure
make
make install
php -m | grep event #看看装上了没
复制代码

<?php

use Event;
use EventBase;

class Reactor
{
    protected $reactor;

    protected $events;

    public static $instance = null;

    const READ = Event::READ | Event::PERSIST;

    const WRITE = Event::WRITE | Event::PERSIST;

    public static function getInstance()
    {
        if (is_null(self::$instance)) {
            self::$instance = new self();
            self::$instance->reactor = new EventBase;
        }

        return self::$instance;
    }

    public function add($fd, $what, $cb, $arg = null)
    {
        switch ($what) {
            case self::READ:
                $event = new Event($this->reactor, $fd, self::READ, $cb, $arg);
                break;
            case self::WRITE:
                $event = new Event($this->reactor, $fd, self::WRITE, $cb, $arg);
                break;
            default:
                $event = new Event($this->reactor, $fd, $what, $cb, $arg);
                break;
        }

        $event->add();
        $this->events[(int) $fd][$what] = $event;
    }

    public function del($fd, $what = 'all')
    {
        $events = $this->events[(int) $fd];
        if ($what == 'all') {
            foreach ($events as $event) {
                $event->free();
            }
        } else {
            if ($what != self::READ && $what != self::WRITE) {
                throw new Exception('不存在的事件');
            }

            $events[$what]->free();
        }
    }

    public function run()
    {
        $this->reactor->loop();
    }

    public function stop()
    {
        foreach ($this->events as $events) {
            foreach ($events as $event) {
                $event->free();
            }
        }
        $this->reactor->stop();
    }
}
复制代码

<?php 

use Throwable;
use MonologHandlerStreamHandler;

class Server
{
	protected $ip;

	protected $port;

	protected $socket;

	protected $reactor;

	public function __construct($ip, $port)
	{
		$this->ip = $ip;
		$this->port = $port;
	}

	public function start()
	{
	    $socket = $this->createTcpConnection();
	    stream_set_blocking($socket, false);

	    Reactor::getInstance()->add($socket, Reactor::READ, function($socket) {
                $conn = stream_socket_accept($socket);
                stream_set_blocking($conn, false);
                (new Connection($conn))->handle();
        });

            Reactor::getInstance()->run();
	}

	public function createTcpConnection()
	{
		$schema = sprintf("tcp://%s:%d", $this->ip, $this->port);
		$socket = stream_socket_server($schema, $errno, $errstr);

		if ($errno) {
			throw new Exception($errstr);
		}

		return $socket;
	}
}
复制代码

<?php

class Connection
{
    protected $conn;

    protected $read_buffer = '';

    protected $write_buffer = '';

    public function __construct($conn)
    {
        $this->conn = $conn;
    }

    public function handle()
    {
        Reactor::getInstance()->add($this->conn, Reactor::READ, Closure::fromCallable([$this, 'read']));
    }

    private function read($conn)
    {
        $this->read_buffer = '';
        if (is_resource($conn)) {
            while ($content = fread($conn, 65535)) {
                $this->read_buffer .= $content;
            }
        }

        if ($this->read_buffer) {
            Reactor::getInstance()->add($conn, Reactor::WRITE, Closure::fromCallable([$this, 'write']));
        } else {
            Reactor::getInstance()->del($conn);
            fclose($conn);
        }
    }

    private function write($conn)
    {
        if (is_resource($conn)) {
            fwrite($conn, "HTTP/1.1 200 OKrnContent-Type: text/html;charset=utf8rnContent-Length:11rnConnection: keep-alivernrnHello!world");
        }
    }

}
复制代码