面向管理后台的系统中，经常会有文件导入的需求。常规的做法就是同步等待，但在业务关系复杂(多表数据校验)、数据量较大的情况下，管理人员只能等结果，也可能会等到超时。
使用异步的话，将导入数据的功能与后端接口解耦，避免接口超时导致的任务中止，也无需前端只为了拿个结果一直保持连接等待。
前端在上传文件后，后端接口将导入任务推送(MQ、管道...)出去，然后直接返回前端。导入服务接到任务执行导入，并根据需求将实时导入状态维护到缓存中。前端查询/轮询后端从缓存取出当前导入状态。
流程图如下：
异步导入.png
简单的PHP + Swoole后端代码示例(实际就两个接口方法upFile、importStatus，和Task的导入处理)：
/**
* Created by PhpStorm.
* User: wen
* Date: 2018/12/8
* Time: 11:09 PM
*/
require 'vendor/autoload.php';
use Swoole\Http\Server;
const BASE_DIR = __DIR__;
// 路由定义
$router = [
'GET' => [
'/importStatus' => 'importStatus' // 查询导入状态
],
'POST' => [
'/upFile' => 'upFile' // 上传导入文件
]
];
// ----SWOOLE-HTTP服务设置部分
$http = new Server("127.0.0.1", 9501);
$http->set([
'worker_num' => 2,
'task_worker_num' => 4,
]);
$http->on('request', function ($request, $response) use ($router, $http) {
$funName = $router[$request->server['request_method']][$request->server['request_uri']] ?? 'NotFound';
if (!function_exists($funName)){
return backJson($response, null, 404, 'ROUTER NOT FOUND');
}
try{
$funName($request, $response, $http);
}catch (Exception $e){
return backJson($response, null, 500, $e->getMessage());
}
});
// 实际导入操作
$http->on('Task', function (swoole_server $serv, $task_id, $from_id, $data) {
$redis = getNewRedis();
$status = [
'step' => 1, // 文件准备处理
'progressRate' => '',
'info' => [],
];
$redis->set($data, json_encode($status, JSON_UNESCAPED_UNICODE));
// 读取文件 使用了PhpOffice\PhpSpreadsheet解析EXCEL
$spreadsheet = \PhpOffice\PhpSpreadsheet\IOFactory::load(BASE_DIR . '/' . $data);
$sheetData = $spreadsheet->getActiveSheet()->toArray(null, true, true, true);
$count = count($sheetData);
$status['step'] = 2; // 文件解析完成
$status['progressRate'] = "解析到文件数据{$count}条";
$redis->set($data, json_encode($status, JSON_UNESCAPED_UNICODE));
$names = [];
var_dump($sheetData);
foreach ($sheetData as $k => $item){
if (1==$k) continue; // 第一行为表头
if (empty($item['A'])) {
unset($sheetData[$k]);
$status['info'][] = "第{$k}行姓名为空";
continue;
}
$names[] = $item['A'];
}
$redis->set($data, json_encode($status, JSON_UNESCAPED_UNICODE));
// TODO: 验证数据库name是否已存在 插入等业务处理...(此处代码省略)
// TODO: 将进度维护到redis
});
$http->on('Finish', function () {});
// ----基础函数部分
function getNewRedis(){
($redis = new \Redis())->connect('127.0.0.1');
return $redis;
}
function backJson($response, $content, $statusCode=200, $msg=''){
$response->header('Content-Type', 'application/json');
$jsonData = [
'statusCode' => $statusCode,
'content' => $content,
'msg' => $msg,
];
$response->end(json_encode($jsonData, JSON_UNESCAPED_UNICODE));
return true;
}
function NotFound($request, $response){
return backJson($response, null, 404, 'ROUTER NOT FOUND');
}
// ----接口方法
// 上传文件
function upFile($request, $response, $server){
$file = $request->files['file'] ?? null;
if (empty($file)) { throw new Exception('未收到上传文件'); }
$importSN = md5($file['tmp_name'] . time()) . '.' . pathinfo($file['name'])['extension'];
$bol = move_uploaded_file($file['tmp_name'], BASE_DIR . '/' . $importSN);
if (false === $bol) { throw new Exception('文件处理异常'); }
$status = [
'step' => '0',
'progressRate' => '',
'info' => [],
];
getNewRedis()->set($importSN, json_encode($status, JSON_UNESCAPED_UNICODE));
$server->task($importSN);
return backJson($response, ['importSN'=>$importSN]);
}
// 查询导入状态
function importStatus($request, $response){
$importSN = $request->get['importSN'] ?? null;
if (!$importSN){ throw new Exception('导入任务编号不正确'); }
$redis = getNewRedis();
$content = $redis->get($importSN);
if (!$content){ throw new Exception('未查询到任务'); }
return backJson($response, json_decode($content));
}
$http->start();
PostMan访问示例：
屏幕快照 2018-12-09 上午4.06.17.png
屏幕快照 2018-12-09 上午4.04.16.png
屏幕快照 2018-12-09 上午4.04.50.png
这里主要是任务投递的渠道，如Channel、MQ服务、Unix Socket等。
Channel：最简单好用，同服务进程内通信，进程挂了就都gg
MQ服务：独立服务，简单通用，可以多服务器，可靠性高
Unix Socket：单服务器内进程间通信，偏底层
