message server – 知识点滴

订阅者注册推送地址，关注的事件
服务端将消息存储redis队列
服务端开N个进程，读取redis队列并推送到订阅者url
推送错误进入错误队列
服务定时发送错误队列消息

原型

ini_set('default_socket_timeout', -1);
$server = new swoole_http_server('0.0.0.0', 8081);

$n = 5;
for($i = 0; $i < $n; $i++){
    $server->addProcess(new swoole_process(function($process) {
        $redis = getRedis();
        while($data = $redis->blPop('test_list', 0)){
            echo $process->pid . " work : " .  $data[1] . "\n";
            sleep(1);
        }
    }));
}

function getRedis(){
    static $redis;
    if(!isset($redis)){
        $redis = new Redis();
        $redis->connect('127.0.0.1', 6379, 0);
        $redis->setOption(Redis::OPT_SERIALIZER, Redis::SERIALIZER_NONE);
    }
    return $redis;
}

$server->on('request', function(swoole_http_request $request, swoole_http_response $response) {
    getRedis()->lPush('test_list', "test info");
    $response->end("<h1>hello swoole. #" . rand(1000, 9999) . "</h1>");
});

$server->start();

ini_set('default_socket_timeout', -1);

$server = new swoole_http_server('0.0.0.0', 8081);

$n = 5;

for($i = 0; $i < $n; $i++){

$server->addProcess(new swoole_process(function($process) {

$redis = getRedis();

while($data = $redis->blPop('test_list', 0)){

echo $process->pid . " work : " . $data[1] . "\n";

sleep(1);

}

}));

}

function getRedis(){

static $redis;

if(!isset($redis)){

$redis = new Redis();

$redis->connect('127.0.0.1', 6379, 0);

$redis->setOption(Redis::OPT_SERIALIZER, Redis::SERIALIZER_NONE);

}

return $redis;

}

$server->on('request', function(swoole_http_request $request, swoole_http_response $response) {

getRedis()->lPush('test_list', "test info");

$response->end("<h1>hello swoole. #" . rand(1000, 9999) . "</h1>");

});

$server->start();

swoole->addProcess(swoole_process)增加的进程不能start()
process结束后会被master进程再次创建，所以要么阻塞或进epoll循环
参考 http://wiki.swoole.com/wiki/page/390.html
与worker或task进程通信参考 http://wiki.swoole.com/wiki/page/363.html

知识点滴

专注于php应用开发

标签： message server

简单的推送消息服务