swoole 学习第一章 Task进程与Timer进程

[pushaowei@localhost ~]# ipcs -m

------ Shared Memory Segments --------
key        shmid      owner      perms      bytes      nattch     status
0x00000000 0          root       600        524288     21         dest

[pushaowei@localhost www]# nc 127.0.0.1 9501

$serv->set(array(
    'task_worker_num' => 8
));

<?php
class Server
{
    private $serv;
    public function __construct() {
        $this->serv = new swoole_server("0.0.0.0", 9501);
        $this->serv->set(array(
            'worker_num' => 8,
            'daemonize' => false,
            'max_request' => 10000,
            'dispatch_mode' => 2,
            'debug_mode'=> 1,
            'task_worker_num' => 8
        ));
        $this->serv->on('Start', array($this, 'onStart'));
        $this->serv->on('Connect', array($this, 'onConnect'));
        $this->serv->on('Receive', array($this, 'onReceive'));
        $this->serv->on('Close', array($this, 'onClose'));
        // bind callback
        $this->serv->on('Task', array($this, 'onTask'));
        $this->serv->on('Finish', array($this, 'onFinish'));
        $this->serv->start();
    }
    public function onStart( $serv ) { //创建连接要经历的
        echo "Start\n";
    }
    public function onConnect( $serv, $fd, $from_id ) { //客户端有连接过来会发起的，有几个连接就有几个fd
        echo "Client {$fd} connect\n";
    }
    public function onReceive( swoole_server $serv, $fd, $from_id, $data ) {
        //客户端发送来的数据就跑到这里来了，fd就是第几个连接上的
        echo "来自客户端的请求 本次的fd为 {$fd}:{$data}\n";
        // 如果我们收到了来自客户端的数据，需要创建一个任务，我们可以通过这样的方式
        //首先我们创建一个数组用于存放需要传递给task的数据
        $param =[
            'task' =>'task_1', //task任务
            'param' => $data, //收到来自客户端的数据传递给它
            'fd' => $fd //客户端的描述符也给它传递过去
        ];
        $serv->task( json_encode( $param ) );
        //通过serv的task方法将数据传递出去，task传递的时候只能传递一个字符串，需要用json或者序列化将这个数据处理
    }
    public function onClose( $serv, $fd, $from_id ) {
        echo "Client {$fd} close connection\n";
    }
    public function onTask($serv,$task_id,$from_id, $data) {

        //onTask回调中会收到onReceive发来的这个任务，
        echo "这次发送过来的task_id 为 {$task_id} 我们的from_id 为 {$from_id}\n";
        $data = json_decode($data,true); //解析发送来的数据
        echo "Receive Task :{$data['task']}\n";
        var_dump($data['param']); //打印客户端发来的数据

        $serv->send( $data['fd'], "hello task task_id == {$task_id}"); //这是发给客户端的

        //处理完成之后调用send函数通过接收到的描述符给客户端发送数据
        return "Task {$task_id}'s result";
        //然后return 返回给worker 进程 告诉他们想说的
    }
    public function onFinish($serv,$task_id, $data) {
        //onFinish收到上面回来的消息就可以将这个data 打印出来
        echo "Task {$task_id} finish\n";
        echo "Result: {$data}\n";
    }
}
$server = new Server();

create database test charset 'utf8';
use test;
 create table vg(
    -> id int unsigned auto_increment,
    -> value char(10) not null default'',
    -> primary key(id)
    -> )engine=InnoDb;

<?php

/**
 1. Class MySqlPlool by:pushaowei
 */
class MySqlPlool{

	private $serv;
	private $pdo;
	public function __construct(){
		$this->serv = new swoole_server('127.0.0.1',9502);
		$this->serv ->set([
			'worker_num' =>8,
			'daemonize' =>false,
			'max_request' =>10000,
			'dispatch_mode' =>3,
			'debug_mode' =>1 ,
			'task_worker_num' =>8
			]);
		$this->serv->on('WorkerStart', array($this, 'onWorkerStart'));
		$this->serv->on('Connect', array($this, 'onConnect'));
		$this->serv->on('Receive', array($this, 'onReceive'));
		$this->serv->on('Close', array($this, 'onClose'));
		// bind callback
		$this->serv->on('Task', array($this, 'onTask'));
		$this->serv->on('Finish', array($this, 'onFinish'));
		$this->serv->start();
	}

	/**
	 * [onConnect 创建连接]
	 * @param  [type] $serv    [description]
	 * @param  [type] $fd      [description]
	 * @param  [type] $from_id [description]
	 * @return [type]          [description]
	 */
	public function onConnect($serv, $fd ,$from_id ){
		echo "Client {$fd} connect \n";
	}

	/**
	 * [onWorkerStart 创建pdo连接,woker进程创建之初被调用]
	 * @param  [type] $serv      [description]
	 * @param  [type] $worker_id [description]
	 * @return [type]            [description]
	 */
	public function onWorkerStart($serv, $worker_id){
		echo "onWorkerStart\n";
		$this->pdo = new PDO("mysql:dbname=test;host=127.0.0.1",'root','123456');
		$this->pdo -> exec('SET NAMES utf8');//设置通信编码
		$this->pdo -> setAttribute(PDO::ATTR_ERRMODE,PDO::ERRMODE_EXCEPTION);

	}
	/**
	 * [onReceive 这里收到客户端的请求]
	 * @param  swoole_server $serv    [description]
	 * @param  [type]        $fd      [description]
	 * @param  [type]        $from_id [description]
	 * @param  [type]        $data    [description]
	 * @return [type]                 [description]
	 */
	public function  onReceive (swoole_server $serv ,$fd , $from_id ,$data ){
		$task= [
			'sql' =>'insert into vg  values (?,?)',
			'params' => ['',$fd],
			'fd' => $fd //描述符
		];
		$serv ->task(json_encode($task));
	}
	/**
	 * [onTask 处理sql,接收客户端的$data]
	 * @param  [type] $serv    [description]
	 * @param  [type] $task_id [description]
	 * @param  [type] $from_id [description]
	 * @param  [type] $data    [description]
	 * @return [type]          [description]
	 */
	public function onTask($serv,$task_id,$from_id,$data){
		echo '已经跑onTask这里来啦';

		try{
			$data = json_decode($data,true);

			$stement = $this->pdo->prepare($data['sql']);
			$stement ->execute($data['params']);
			$serv->send($data['fd'],'insert succed'); //将返回结果给客户端
			return true;
		}catch(PDOException $e)
		{
			var_dump($e->getMessage());
			return false;
		}
	}
	/**
	 * [onFinish description]
	 * @param  [type] $serv    [description]
	 * @param  [type] $task_id [description]
	 * @param  [type] $data    [description]
	 * @return [type]          [description]
	 */
	public function onFinish($serv,$task_id,$data){
		var_dump("resut:" .$data);
	}

	/**
	 * @param $serv
	 * @param $fd
	 * @param $from_id
	 */
	public function onClose( $serv, $fd, $from_id ) {
		echo "Client {$fd} close connection \n";
	}
}
new MySqlPlool();

public function onWorkerStart($serv, $worker_id){
	echo "onWorkerStart\n";
	//
	//让它只有是taskworker的时候才创建pdo连接
	if($serv -> taskworker){
		$this->pdo = new PDO("mysql:dbname=test;host=127.0.0.1",'root','123456');
		$this->pdo -> exec('SET NAMES utf8');//设置通信编码
		$this->pdo -> setAttribute(PDO::ATTR_ERRMODE,PDO::ERRMODE_EXCEPTION);
	}
	else
	 echo "我们只让taskworker连";
}

[pushaowei@localhost www]# php server.php
我们只让task worker 连接
我们只让task worker 连接
我们只让task worker 连接
我们只让task worker 连接
我们只让task worker 连接
我们只让task worker 连接
taskWorkerStart
taskWorkerStart
我们只让task worker 连接
taskWorkerStart
我们只让task worker 连接
taskWorkerStart
taskWorkerStart
taskWorkerStart
taskWorkerStart
taskWorkerStart

$this->serv ->set([
	'worker_num' =>8,
	'daemonize' =>false,
	'max_request' =>10000,
	'dispatch_mode' =>3,
	'debug_mode' =>1 ,
	'task_worker_num' =>8
	]);

int swoole_timer_tick(int $ms, mixed $callback, mixed $param = null);
int swoole_timer_after(int $after_time_ms, mixed $callback_function);

public function onWorkerStart($serv, $worker_id){
	//当worker id =0 的时候我们才创建这个tick
	if($worker_id == 0)
	{
		swoole_timer_tick(1000, function($timer_id,$parmas){
			echo "QQ:542684913\n";
			echo "{$parmas} \n";
		}, "hello");
	}
}

?php
/**
 * Created by PhpStorm.
 * User: ClownFish 542684913@qq.com
 * Date: 14-12-27
 * Time: 上午11:59
 */
class ParseCrontab
{
    static public $error;
    /**
     *  解析crontab的定时格式，linux只支持到分钟/，这个类支持到秒
     * @param string $crontab_string :
     *
     *      0     1    2    3    4    5
     *      *     *    *    *    *    *
     *      -     -    -    -    -    -
     *      |     |    |    |    |    |
     *      |     |    |    |    |    +----- day of week (0 - 6) (Sunday=0)
     *      |     |    |    |    +----- month (1 - 12)
     *      |     |    |    +------- day of month (1 - 31)
     *      |     |    +--------- hour (0 - 23)
     *      |     +----------- min (0 - 59)
     *      +------------- sec (0-59)
     * @param int $start_time timestamp [default=current timestamp]
     * @return int unix timestamp - 下一分钟内执行是否需要执行任务，如果需要，则把需要在那几秒执行返回
     * @throws InvalidArgumentException 错误信息
     */
    static public function parse($crontab_string, $start_time = null)
    {
        if (is_array($crontab_string)) {
            return self::_parse_array($crontab_string, $start_time);
        }
        if (!preg_match('/^((\*(\/[0-9]+)?)|[0-9\-\,\/]+)\s+((\*(\/[0-9]+)?)|[0-9\-\,\/]+)\s+((\*(\/[0-9]+)?)|[0-9\-\,\/]+)\s+((\*(\/[0-9]+)?)|[0-9\-\,\/]+)\s+((\*(\/[0-9]+)?)|[0-9\-\,\/]+)\s+((\*(\/[0-9]+)?)|[0-9\-\,\/]+)$/i', trim($crontab_string))) {
            if (!preg_match('/^((\*(\/[0-9]+)?)|[0-9\-\,\/]+)\s+((\*(\/[0-9]+)?)|[0-9\-\,\/]+)\s+((\*(\/[0-9]+)?)|[0-9\-\,\/]+)\s+((\*(\/[0-9]+)?)|[0-9\-\,\/]+)\s+((\*(\/[0-9]+)?)|[0-9\-\,\/]+)$/i', trim($crontab_string))) {
                self::$error = "Invalid cron string: " . $crontab_string;
                return false;
            }
        }
        if ($start_time && !is_numeric($start_time)) {
            self::$error = "\$start_time must be a valid unix timestamp ($start_time given)";
            return false;
        }
        $cron = preg_split("/[\s]+/i", trim($crontab_string));
        $start = empty($start_time) ? time() : $start_time;
        if (count($cron) == 6) {
            $date = array(
                'second' => self::_parse_cron_number($cron[0], 0, 59),
                'minutes' => self::_parse_cron_number($cron[1], 0, 59),
                'hours' => self::_parse_cron_number($cron[2], 0, 23),
                'day' => self::_parse_cron_number($cron[3], 1, 31),
                'month' => self::_parse_cron_number($cron[4], 1, 12),
                'week' => self::_parse_cron_number($cron[5], 0, 6),
            );
        } elseif (count($cron) == 5) {
            $date = array(
                'second' => array(1 => 1),
                'minutes' => self::_parse_cron_number($cron[0], 0, 59),
                'hours' => self::_parse_cron_number($cron[1], 0, 23),
                'day' => self::_parse_cron_number($cron[2], 1, 31),
                'month' => self::_parse_cron_number($cron[3], 1, 12),
                'week' => self::_parse_cron_number($cron[4], 0, 6),
            );
        }
        if (
            in_array(intval(date('i', $start)), $date['minutes']) &&
            in_array(intval(date('G', $start)), $date['hours']) &&
            in_array(intval(date('j', $start)), $date['day']) &&
            in_array(intval(date('w', $start)), $date['week']) &&
            in_array(intval(date('n', $start)), $date['month'])
        ) {
            return $date['second'];
        }
        return null;
    }
    /**
     * 解析单个配置的含义
     * @param $s
     * @param $min
     * @param $max
     * @return array
     */
    static protected function _parse_cron_number($s, $min, $max)
    {
        $result = array();
        $v1 = explode(",", $s);
        foreach ($v1 as $v2) {
            $v3 = explode("/", $v2);
            $step = empty($v3[1]) ? 1 : $v3[1];
            $v4 = explode("-", $v3[0]);
            $_min = count($v4) == 2 ? $v4[0] : ($v3[0] == "*" ? $min : $v3[0]);
            $_max = count($v4) == 2 ? $v4[1] : ($v3[0] == "*" ? $max : $v3[0]);
            for ($i = $_min; $i <= $_max; $i += $step) {
                $result[$i] = intval($i);
            }
        }
        ksort($result);
        return $result;
    }
    static protected function _parse_array($crontab_array, $start_time)
    {
        $result = array();
        foreach ($crontab_array as $val) {
            if(count(explode(":",$val)) == 2){
                $val = $val.":01";
            }
            $time = strtotime($val);
            if ($time >= $start_time && $time < $start_time + 60) {
                $result[$time] = $time;
            }
        }
        return $result;
    }
}