這篇文章帶給大家的內容是關於php如何使用SwooleTaskWorker實現非同步操作Mysql（程式碼），有一定的參考價值，有需要的朋友可以參考一下，希望對你有幫助。
在一般的 Server 程式中都會有一些耗時的任務，例如：發送郵件、聊天伺服器發送廣播等。如果我們採用同步阻塞的防水去執行這些任務，那麼這肯定會非常的慢。
Swoole 的 TaskWorker 進程池可以用來執行一些非同步的任務，而且不會影響接下來的任務，很適合處理以上場景。
那麼什麼是非同步任務呢？
可以從下面的圖示來簡單了解一下。 （來自網絡，侵刪）
我們上一個Swoole 的文章介紹瞭如何創建一個簡單的伺服器，並且知道了幾個核心的回調函數的使用方法。
要實現上述的非同步處理，只需要增加兩個事件回呼即可：onTask 和 onFinish, 這兩個回呼函數分別用於執行 Task 任務和處理 Task 任務的回傳結果。另外還需要在 set 方法中設定 task 進程數量。
使用範例：
class Server
{
private $serv;
public function __construct() {
$this->serv = new swoole_server("0.0.0.0", 9501);
$this->serv->set(array(
'worker_num' => 4,
'daemonize' => false,
'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'));
$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";
// 发送任务到Task进程
$param = array(
'fd' => $fd
);
$serv->task( json_encode( $param ) );
echo "继续处理之后的逻辑\n";
}
public function onTask($serv, $task_id, $from_id, $data) {
echo "This Task {$task_id} from Worker {$from_id}\n";
echo "Data: {$data}\n";
for($i = 0 ; $i send( $fd , "Data in Task {$task_id}");
return "Task {$task_id}'s result";
}
public function onFinish($serv,$task_id, $data) {
echo "Task {$task_id} finish\n";
echo "Result: {$data}\n";
}
public function onStart( $serv ) {
echo "Server Start\n";
}
public function onConnect( $serv, $fd, $from_id ) {
echo "Client {$fd} connect\n";
}
public function onClose( $serv, $fd, $from_id ) {
echo "Client {$fd} close connection\n";
}
}
$server = new Server();透過上述範例可以看到，發起一個非同步任務只需要呼叫 swoole_server 的 task 方法就可以。發送之後會觸發 onTask 回調，可以透過 $task_id 和 $from_id 處理不同進程的不同任務。最後可以透過 return 一個字串來將執行結果傳回 Worker 進程，Worker 進程透過 onFinish 回呼來處理結果。
那麼基於上述程式碼就可以實現非同步操作 mysql。非同步操作mysql 較適合以下場景：
並發的讀寫操作
沒有時序上的嚴格關係
#不影響主執行緒邏輯
好處：
#提高並發
降低IO 消耗
資料庫的壓力主要在於mysql 維持的連接數，如果存在1000 個並發，那麼mysql 就需要建立對應數量的連結。而採用長連線的方式，mysql 的連線一直維持在進程中，減少了建立連線的損耗。可以透過 swoole 開啟多個 task 進程，每一個進程內維持一個mysql 長連接，那麼這樣子也可以引申出來 mysql 連接池技術。另外要注意的是，mysql 伺服器如果偵測到長時間沒有沒有查詢，則會中斷連線回收資源,所以要有斷線重連的機制。
以下是一個簡單的非同步操作 mysql 的範例：
或以上的程式碼，我們只需要修改 onReceive、onTask、onFinish 三個函數。
class Server
{
private $serv;
public function __construct() {
$this->serv = new swoole_server("0.0.0.0", 9501);
$this->serv->set(array(
'worker_num' => 4,
'daemonize' => false,
'task_worker_num' => 8 // task进程数量 即为维持的MySQL连接的数量
));
$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'));
$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 "收到数据". $data . PHP_EOL;
// 发送任务到Task进程
$param = array(
'sql' => $data, // 接收客户端发送的 sql
'fd' => $fd
);
$serv->task( json_encode( $param ) ); // 向 task 投递任务
echo "继续处理之后的逻辑\n";
}
public function onTask($serv, $task_id, $from_id, $data) {
echo "This Task {$task_id} from Worker {$from_id}\n";
echo "recv SQL: {$data['sql']}\n";
static $link = null;
$sql = $data['sql'];
$fd = $data['fd'];
HELL:
if ($link == null) {
$link = @mysqli_connect("127.0.0.1", "root", "root", "test");
}
$result = $link->query($sql);
if (!$result) { //如果查询失败
if(in_array(mysqli_errno($link), [2013, 2006])){
//错误码为2013，或者2006，则重连数据库，重新执行sql
$link = null;
goto HELL;
}
}
if(preg_match("/^select/i", $sql)){//如果是select操作，就返回关联数组
$data = array();
while ($fetchResult = mysqli_fetch_assoc($result) ){
$data['data'][] = $fetchResult;
}
}else{//否则直接返回结果
$data['data'] = $result;
}
$data['status'] = "OK";
$data['fd'] = $fd;
$serv->finish(json_encode($data));
}
public function onFinish($serv, $task_id, $data) {
echo "Task {$task_id} finish\n";
$result = json_decode($result, true);
if ($result['status'] == 'OK') {
$this->serv->send($result['fd'], json_encode($result['data']) . "\n");
} else {
$this->serv->send($result['fd'], $result);
}
}
public function onStart( $serv ) {
echo "Server Start\n";
}
public function onConnect( $serv, $fd, $from_id ) {
echo "Client {$fd} connect\n";
}
public function onClose( $serv, $fd, $from_id ) {
echo "Client {$fd} close connection\n";
}
}
$server = new Server();以上程式碼在 onReceive 時直接接收一條 sql，之後直接傳送到 Task 任務中。這時候下一步的流程緊接著輸出，這裡也體現出了非同步。然後 onTask 和 onFinish 分別用來傳送 sql，處理 task 執行結果給資料庫。
#以上是php如何使用SwooleTaskWorker實作非同步操作Mysql（程式碼）的詳細內容。更多資訊請關注PHP中文網其他相關文章！
