php安装gearman扩展实现异步分步式任务

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

$client=newGearmanClient();

$client->addServer('127.0.0.1', 4730);//本机可以直接addServer(),默认服务器端使用4730端口

$client->setCompleteCallback('completeCallBack');//先绑定才有效

$result1=$client->do('say','do');//do是同步进行，进行处理并返回处理结果。

$result2=$client->doBackground('say','doBackground');//异步进行，只返回处理句柄。

$result3=$client->addTask('say','addTask');//添加任务到队列,同步进行？通过添加task可以设置回调函数。

$result4=$client->addTaskBackground('say','addTaskBackground');//添加后台任务到队列，异步进行？

$client->runTasks();//运行队列中的任务，只是do系列不需要runTask()。

echo'result1:';

var_dump($result1);

echo'
';

echo'result2:';

var_dump($result2);

echo'
';

echo'result3:';

var_dump($result3);

echo'
';

echo'result4:';

var_dump($result4);

echo'
';

//绑定回调函数，只对addTask有效

functioncompleteCallBack($task)

{

    echo'CompleteCallback！handle result:'.$task->data().'
';

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

$worker=newGearmanWorker();

$worker->addServer();

$worker->addFunction('say',function(GearmanJob$job){

    $workload=$job->workload();//接收client传递的数据

    echo'receive data:'.$workload.PHP_EOL;

    returnstrrev($workload);//仅作反转处理

});

//无际循环运行，gearman内部已有处理，不会出现占用过高死掉的情况

while($worker->work()){

    if($worker->returnCode() !== GEARMAN_SUCCESS){

        echo'error'.PHP_EOL;

    }

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

declare(ticks = 1);

$num= 5;//最大子进程数

$child= 0;//当前子进程数

//信号处理函数

functionsig_handler($sig)

{

    global$child;

    switch($sig) {

        caseSIGCHLD:

            $child--;

            echo'SIGCHLD received! now we have '.$child.' process'.PHP_EOL;

            break;

        caseSIGINT:

            $child--;

            echo'SIGINT received! now we have '.$child.' process'.PHP_EOL;

            break;

        caseSIGTERM:

            $child--;

            echo'SIGTERM received! now we have '.$child.' process'.PHP_EOL;   

            break;

        default:

            # code...

            break;

    }

}

//安装信号处理器

pcntl_signal(SIGTERM,"sig_handler");//进程被kill时发出的信号

// pcntl_signal(SIGHUP,  "sig_handler");//终端关闭时发出的信号

pcntl_signal(SIGINT,"sig_handler");//中断进程信号，如Ctrl+C

pcntl_signal(SIGCHLD,"sig_handler");//进程退出信号

while(true)

{

    $child++;

    $parentpid=getmypid();

    $pid= pcntl_fork();//一分为二，父进程和子进程都会执行以下代码

    if($pid== -1)

    {

        exit("can not fork!");//出错

    }elseif($pid> 0){

        //父进程处理代码

        echo'I am parent.my pid is'.$pid.' and my parent pid is'.$parentpid.PHP_EOL;

        if($child>=$num)

        {

            pcntl_wait($status);//挂起，while语句不会继续执行。等待子进程结束，防止子进程成为僵尸进程

        }

    }elseif($pid== 0){

        //子进程代码

        echo'I am child, and my parent pid is '.$parentpid." my pid is ".getmypid()." now have $child process".PHP_EOL;       

        //执行具体代码

        pcntl_exec('/usr/bin/php',array('/var/www/test/mywork.php'));

    }

    pcntl_signal_dispatch();//分发信号，使安装的信号处理器能接收。

    //低于php5.3该函数无效，但有开头的declare (ticks = 1);表示每执行一条低级指令，

    //就检查一次信号，如果检测到注册的信号，就调用其信号处理器

    sleep(rand(3,5));//防止100%占用

}