关于异步任务队列
用户打开了我们的网站。他要做的就是勾选需要发邮件的代理商列表，然后把结算邮件发出去。
假如我们需要发 1 封邮件，我们写个函数执行即可。
考虑到网络可能会稍微有点延迟，但是是可以接受的，用户会乖乖等你的网页发完邮件了再关闭网页。
假如我们要发布 10 封邮件，用一个 for 循环，循环 10 遍执行发邮件操作。
这时候，也许 10 倍的网络延迟会让用户稍微有点不耐烦，但勉强可以等吧。
假如要发 100 封邮件，for 循环 100 遍，用户直接揭竿而起，什么破网站！
但实际上，我们很可能有超过 1 万的邮件。怎么处理这个延迟的问题？
答案就是用异步。
把“发邮件”这个操作封装，然后后台异步地执行 1 万遍。
这样的话，用户提交网页后，他所等待的时间只是“把发邮件任务请求推送进队列里”的时间。
而我们的后台服务将在用户看不见的地方跑。
在实现“异步队列”这点上，有人采用 mysq l 表或者 redis 来存放待发送的邮件，然后，每分钟定时读取待发送列表，然后处理。这便是定时异步任务队列。但当前提交的任务要一分钟后才能执行，在某些实时性要求应用场景里还是不快。有些场景要求，只有一提交任务，便马上执行，但用户不需要等待返回结果。
在云平台 SAE 和 BA E 上，都有 taskqueue 服务来解决上面的问题。而如果是自己假设服务器，则如何解决？本文将探讨用 php 扩展 swoole 实现实时异步任务队列的方案。
安装 swoole
pecl 安装:
pecl install swoole
看命令行提示，如果它提示说没有写 php.ini，则自己手动在 php.ini 后面加上：
extension = "swoole.so"
当然你也可以编译源码进行安装，在此就不多讲了。
服务端
在打算放置脚本的目录（你也可以自行新建）新建 Server.php，代码如下：
<?php
class Server
{
private $serv;
public function __construct()
{
$this->serv = new swoole_server("0.0.0.0", 9501);
$this->serv->set(array(
'worker_num' => 1, //一般设置为服务器 CPU 数的 1-4 倍
'daemonize' => 1, //以守护进程执行
'max_request' => 10000,
'dispatch_mode' => 2,
'task_worker_num' => 8, //task 进程的数量
"task_ipc_mode " => 3, //使用消息队列通信，并设置为争抢模式
//"log_file" => "log/taskqueueu.log" ,//日志
));
$this->serv->on('Receive', array($this, 'onReceive'));
// bind callback
$this->serv->on('Task', array($this, 'onTask'));
$this->serv->on('Finish', array($this, 'onFinish'));
$this->serv->start();
}
public function onReceive(swoole_server $serv, $fd, $from_id, $data)
{
//echo "Get Message From Client {$fd}:{$data}\n";
// send a task to task worker.
$serv->task($data);
}
public function onTask($serv, $task_id, $from_id, $data)
{
$array = json_decode($data, true);
if ($array['url']) {
return $this->httpGet($array['url'], $array['param']);
}
}
public function onFinish($serv, $task_id, $data)
{
//echo "Task {$task_id} finish\n";
//echo "Result: {$data}\n";
}
protected function httpGet($url, $data)
{
if ($data) {
$url .= '?' . http_build_query($data);
}
$curlObj = curl_init(); //初始化 curl，
curl_setopt($curlObj, CURLOPT_URL, $url); //设置网址
curl_setopt($curlObj, CURLOPT_RETURNTRANSFER, 1); //将 curl_exec 的结果返回
curl_setopt($curlObj, CURLOPT_SSL_VERIFYPEER, FALSE);
curl_setopt($curlObj, CURLOPT_SSL_VERIFYHOST, FALSE);
curl_setopt($curlObj, CURLOPT_HEADER, 0); //是否输出返回头信息
$response = curl_exec($curlObj); //执行
curl_close($curlObj); //关闭会话
return $response;
}
}
$server = new Server();
由于服务端是异步、常驻内存的，因此必须通过命令行(即“`cli“)来启动。在命令行执行以上代码以启动服务：
php Server.php
执行完毕后关闭命令行窗口即可。服务会在后台以守护进程运行
客户端
启动服务后，让我们看看如何调用服务。新建测试文件 Client_test.php
代码如下：
<?php
class Client
{
private $client;
public function __construct()
{
$this->client = new swoole_client(SWOOLE_SOCK_TCP);
}
public function connect()
{
if (!$this->client->connect("127.0.0.1", 9501, 1)) { # 后端 Server.php 中设置的 ip 和端口
throw new Exception(sprintf('Swoole Error: %s', $this->client->errCode));
}
}
public function send($data)
{
if ($this->client->isConnected()) {
if (!is_string($data)) {
$data = json_encode($data);
}
return $this->client->send($data);
} else {
throw new Exception('Swoole Server does not connected.');
}
}
public function close()
{
$this->client->close();
}
}
$data = array(
"url" => "http://192.168.1.112/send_mail", # 这里为要执行任务的方法 比如发送邮件
"param" => array(
"username" => 'test',
"password" => 'test'
)
);
$client = new Client();
$client->connect();
if ($client->send($data)) {
echo 'success';
} else {
echo 'fail';
}
$client->close();
在上面代码中，url 即为任务所在地址，param 为所需传递参数。
保存好代码，在命令行或者浏览器中执行 Client_test.php,便实现了异步任务队列。你所填写的 URL，将会在每次异步任务被提交后，以 HTTP GET 的方式异步执行。
查看与关闭
swoole 好像没有很便捷的关闭方式。所以只能直接通过关闭进程来关闭。
查看命令：
ps -ef | grep php
结束单个进程：
kill -9 {进程号}
结束所有进程的命令：
killall -9 php
转载自： http://blog.star7th.com/2016/01/1905.html
