PHP實現MySQL並發查詢

一般的，一個看似很簡單的頁面，一次http請求后，到達服務端，穿過Cache層，落到后台后，實際可能會有很多很多的數據查詢邏輯！而這些查詢實際是不相互依賴的，也即可以同時查詢。比如各種用戶信息，用戶的APP列表，每個APP對應的流量數據、消耗記錄、服務狀態，平台運行狀態，消息通知，新聞資訊等等。
這篇文章主要介紹了數據查詢層，如何把串行變並行，提高查詢效率、提升應用性能。實現方式包括：mysqlnd異步查詢，cURL並發請求，Swoole異步非阻塞！

PHP腳本是按文檔流的形式來執行的，所以我們在編寫PHP程序時，代碼基本都是串行的，尤其是SQL，比如：
<ignore_js_op> 
這種方式，是每次查詢都需要等待結果返回之后再開始下一次的查詢，
運行時間 = （第1次發送請求時間 + 0.01 + 第1次返回時間）+（第2次發送請求時間 + 0.05 + 第2次返回時間）+（第3次發送請求時間 + 0.03 + 第3次返回時間）

如果是並發查詢，那么流程就成了：
<ignore_js_op> 
運行時間 = 發送請求時間 + 0.05 + 返回時間

顯然，並發查詢要比串行查詢快！

那么PHP可以實現並發查詢嗎？答案是肯定的！

一、利用MySQL的異步查詢功能

目前 MySQL 的異步查詢，只在 MySQLi 擴展提供，查詢方法分別是：

• 使用 MYSQLI_ASYNC 模式執行 mysqli::query
• 獲取異步查詢結果：mysqli::reap_async_query

需要注意的是，使用異步查詢，需要使用 mysqlnd 作為PHP的MySQL數據庫驅動，
mysqlnd(MySQL Native Driver) 是 Zend 公司開發的 MySQL 數據庫驅動，采用PHP開源協議，用於代替舊版的由 MySQL AB公司（現在的Oracle）開發的 libmysql，PHP 5.3 及以上版本開始提供，PHP5.4 之后的版本 mysqlnd 為默認配置選項，
如果 PHP 小於 5.4，編譯時需要指定編譯參數：

1. --with-mysqli=mysqlnd --with-pdo-mysql=mysqlnd

復制代碼

MySQL異步查詢示例腳本：

1. /**
2. * MySQL異步查詢示例腳本：
3. * @filename p_async.php
4. * @url http://test.979137.com/ParallelSQL/p_async.php
5. */
7. //期望結果集：獲取以下用戶的每個月每個APP的消費統計
8. $top = array('979137', '555555', '666666', '888888', '999999');
9. $ret = array_fill_keys($top, array());
11. //組織結構查詢
12. $cmd = $resources = array();
13. $sql = "SELECT access_key,SUM(amount) sum_amount FROM consume_2016%s WHERE uid=%d AND product='SAE' GROUP BY access_key"; 
14. foreach($top as $uid) {
15.     for($i = 1; $i <= 12; $i++) {
16.         $ret[$uid][$i] = $tmp = array();
17.         $tmp['uid'] = $uid;
18.         $tmp['month'] = $i; 
19.         $tmp['resource'] = $resources[] = new \mysqli('localhost', 'root', '123456', 'sae', '3306');
20.         $tmp['resource']->set_charset('utf8');
21.         $tmp['resource']->query(sprintf($sql, sprintf('%02d', $i), $uid), MYSQLI_ASYNC);
22.         $tag = spl_object_hash($tmp['resource']);
23.         $cmd[$tag] = $tmp;
24.     }   
25. }
27. $total = $query_times = count($resources);
29. //獲取結果
30. do {
31.     $read = $error = $reject = $resources;
32.     //等待查詢結束
33.     if (!\mysqli::poll($read, $error, $reject, 1)) {
34.         continue;
35.     }   
36.     //批量獲取結果
37.     foreach($read as $resource) {
38.         $result = $resource->reap_async_query();
39.         if ($result) {
40.             $tag = spl_object_hash($resource);
41.             $uid = $cmd[$tag]['uid'];
42.             $month = $cmd[$tag]['month'];
43.             while(($row = $result->fetch_assoc()) != false) {
44.                 $ret[$uid][$month][$row['access_key']] = $row['sum_amount'];
45.             }   
46.             $result->free();
47.             $total--;
48.     
49.         } else die('MySQLi error: '.$resource->error);
50.     }   
51. } while ($total > 0);
53. var_dump($ret);

復制代碼

二、cURL實現並發請求

先解釋下 cURL 和 SQL 怎么就扯上關系了呢！

我們知道在很多系統架構里，PHP是不會直接操作DB的，而是 RESTFull 架構，這時候所有操作都接口化了，
這時上述所講的 SQL 就演變成 接口調用 了，
因為 API，所以 cURL！

以下是PHP中cURL多線程相關函數：

curl_multi_add_handle — 向curl批處理會話中添加單獨的curl句柄
curl_multi_close — 關閉一組cURL句柄
curl_multi_exec — 運行當前 cURL 句柄的子連接
curl_multi_getcontent — 如果設置了CURLOPT_RETURNTRANSFER，則返回獲取的輸出的文本流
curl_multi_info_read — 獲取當前解析的cURL的相關傳輸信息
curl_multi_init — 返回一個新cURL批處理句柄
curl_multi_remove_handle — 移除curl批處理句柄資源中的某個句柄資源
curl_multi_select — 等待所有cURL批處理中的活動連接
curl_multi_setopt — 為 cURL 並行處理設置一個選項
curl_multi_strerror —  返回描述錯誤碼的字符串文本

我們可以利用這些多線程函數，實現 cURL 並發請求，從而實現並發 SQL！

cURL並發請求，服務端接口示例腳本：

1. /**
2. * cURL並發請求，服務端接口示例腳本
3. * @filename consume.php
4. * @url http://test.979137.com/ParallelSQL/consume.php
5. */
6. $resource = new \mysqli('localhost', 'root', '123456', 'sae', '3306');
7. $resource->set_charset('utf8');
9. $sql = "SELECT access_key,SUM(amount) sum_amount FROM consume_%d WHERE uid=%d AND product='%s' GROUP BY access_key";
10. $sql = sprintf($sql, $_GET['ym'], $_GET['uid'], $_GET['product']);
11. $res = $resource->query($sql);
13. $out['code'] = $resource->errno;
14. $out['message'] = $resource->error;
15. $data = array();
16. if (is_object($res)) {
17.     while(($row = $res->fetch_assoc()) != false) {
18.         $data[$row['access_key']] = $row['sum_amount'];
19.     }   
20. }
21. $out['data'] = $data;
23. header('Content-Type: application/json; charset=utf-8');
24. echo json_encode($out);

復制代碼

cURL並發請求，客戶端調用示例腳本：

1. /**
2. * cURL並發請求，客戶端調用示例腳本
3. * @filename p_curl.php
4. * @url http://test.979137.com/ParallelSQL/p_curl.php
5. */
7. //期望結果集：獲取以下用戶的每個月每個APP的消費統計
8. $top = array('979137', '555555', '666666', '888888', '999999');
9. $ret = array_fill_keys($top, array());
11. $mch = curl_multi_init();
12. $opt[CURLOPT_HEADER] = 0;
13. $opt[CURLOPT_CONNECTTIMEOUT] = 60; 
14. $opt[CURLOPT_RETURNTRANSFER] = true;
15. $opt[CURLOPT_HTTPHEADER] = array('Host: api.979137.com');
17. //生成句柄並加入到批處理
18. $cmd = array();
19. foreach($top as $uid) {
20.     for($i = 1; $i <= 12; $i++) {
21.         $ret[$uid][$i] = $tmp = array();
22.         $tmp['url'] = sprintf('http://127.0.0.1/ParallelSQL/consume.php?ym=2016%02d&uid=%d&product=SAE', $i, $uid);
23.         $tmp['uid'] = $uid;
24.         $tmp['month'] = $i; 
25.         $tmp['resource'] = curl_init($tmp['url']);   
26.         curl_setopt_array($tmp['resource'], $opt);
27.         curl_multi_add_handle($mch, $tmp['resource']);
28.         $cmd[] = $tmp;
29.     }   
30. }
32. //並發執行，直到全部結束。
33. do {
34.     curl_multi_exec($mch, $active);   
35. } while ($active > 0); 
37. //獲取全部結果
38. foreach($cmd as $c) {
39.     $res = curl_multi_getcontent($c['resource']);
40.     $http_code = curl_getinfo($c['resource'], CURLINFO_HTTP_CODE);
41.     if ($res === false || $http_code != 200) {
42.         die(curl_error($c['resource']));
43.     }   
44.     $res = json_decode($res, true);
45.     $res['code'] && die($res['message']);
46.     $ret[$c['uid']][$c['month']] = $res['data'];
47. }
48. curl_multi_close($mch);
50. var_dump($ret);

復制代碼

查詢效率對比
1、數據表機構：

1. CREATE TABLE `consume_201612` (
2.   `id` int(10) unsigned NOT NULL AUTO_INCREMENT,
3.   `uid` int(10) unsigned NOT NULL,
4.   `product` enum('UID','SAE','SC2','SCE','SEM','SCS','SLS') NOT NULL DEFAULT 'SAE',
5.   `access_key` varchar(255) NOT NULL,
6.   `service_code` varchar(255) NOT NULL,
7.   `amount` int(10) unsigned NOT NULL,
8.   `remark` varchar(255) NOT NULL,
9.   `data` text NOT NULL,
10.   `times` int(10) unsigned NOT NULL,
11.   PRIMARY KEY (`id`),
12.   KEY `uid_pa` (`uid`,`product`,`access_key`)
13. ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;

復制代碼

2、數據量：總共分12張表，每張表的數量在 500萬~1100萬 之間，全量數據在1億以上

3、MySQL異步查詢結果：

$ mysql.server restart
$ php p_async.php 
Query times: 60
Run time: 2.4453s
(CPU)User time: 0.087966s
(CPU)System time: 0.117625s

4、cURL並發請求查詢結果

$ mysql.server restart
$ php p_curl.php
Query times: 60
Run time: 2.173035s
(CPU)User time: 0.40652s
(CPU)System time: 0.869902s

5、普通串行查詢結果

$ mysql.server restart
$ php p_sync.php 
Query times: 60
Run time: 20.485623s
(CPU)User time: 0.083185s
(CPU)System time: 0.036566s
Memory usage: 304.72kb

並發執行時，我們可以在 MySQL 服務器看到所有的正在執行的SQL：
<ignore_js_op>

WechatIMG81.jpeg (690.09 KB, 下載次數: 0)

下載附件

2016-12-6 18:51 上傳

 

總結

1、在並發查詢下，查詢效率提高了近10倍
2、使用 MySQL 異步查詢，因為需要給所有查詢都創建一個新的連接，而 MySQL 服務端會為每個連接創建一個單獨的線程進行處理，如果創建的線程數過多，會給系統造成負擔，請謹慎使用
3、使用 cURL 並發請求后端接口時，CPU負載明顯上升，所以並發請求后端接口，一定程度上會增加后端壓力，這和前端大流量下的高並發原理是一樣的
4、使用 cURL 並發請求，還需要考慮一個網絡延時的問題，網絡延時越小，查詢效率提升越明顯。如果你是想代替類方法或函數調用，在條件允許的情況，建議直接連接服務器本機即127.0.0.1
5、在並發請求下，因為需要一次性接收全部返回結果，所以會占用更多的內存資源

需要說明的是，在實際應用中 cURL 的並發請求，一般不只單用於數據查詢，而是為了完成更多的后台業務邏輯，
所以，在服務器負載能力允許的情況下，推薦使用 cURL 並行轉發的形式，提升前端響應速度！

最后說一下 Swoole，

Swoole 是 PHP 的異步、並行、高性能網絡通信引擎，使用純C語言編寫，提供了PHP語言的異步多線程服務器，異步TCP/UDP網絡客戶端，異步MySQL，異步Redis，數據庫連接池，AsyncTask，消息隊列，毫秒定時器，異步文件讀寫，異步DNS查詢！
其中的異步MySQL，其原理是通過 MYSQLI_ASYNC 模式查詢，然后獲取 mysql 連接的 socket，加入到 epoll 時間循環中，當數據庫返回結果時會回調指定函數。
這個過程是完全異步非阻塞的，不浪費CPU，具體實現方式這里不再詳細介紹！