01. 背景

最近项目需求里有个文件上传功能，而客户需求里的文件基本上是比较大的，基本上得有 1 GiB 以上的大小，而上传大文件尤其是读大文件，可能会造成卡 UI 或者说点不动的问题。而用后台的 Worker 去实现是一个比较不错的解决办法。

02. 原理讲解

02.01. Shared Worker

Shared Worker 的好处是可以从几个浏览上下文中访问，例如几个窗口、iframe 或其他 worker。这样我们可以保证全局的页面上传任务都在我们的控制之下，甚至可以防止重复提交等功能。

02.02. 组合式函数

组合式函数的好处是在 Vue 3 是可以在任何 *.vue 文件中使用，并且是响应式方法，可以侦听 pinia 内 token 等的变化，传递给 Worker

02.03 简单流程设计

03. 代码

upload-worker.ts 代码

import { sha256 } from '@noble/hashes/sha256'; import { bytesToHex as toHex } from '@noble/hashes/utils'; interface SharedWorkerGlobalScope {   onconnect: (event: MessageEvent<any>) => void; } const _self: SharedWorkerGlobalScope = self as any; /**  * 分片大小  */ const pieceSize = 1024 * 1024; /**  * 消息参数  */ interface MessageArg<T> {   /**    * 函数名    */   func: string;   /**    * 参数    */   arg: T; } /**  * 上传任务信息  */ interface UploadTaskInfo {   /**    * 文件名    */   fileName: string;   /**    * 上传路径    */   uploadPath: string;   /**    * 任务 id    */   id: string;   /**    * 文件大小    */   size: number;   /**    * 上传进度    */   progress: number;   /**    * 上传速度    */   speed?: number;   /**    * 任务状态    */   status: 'uploading' | 'paused' | 'canceled' | 'done' | 'error' | 'waiting';   /**    * 开始时间    */   startTime?: Date;   /**    * 结束时间    */   endTime?: Date;   /**    * 错误信息    */   errorMessage?: string; } /**  * 上传任务  */ interface UploadTask extends UploadTaskInfo {   file: File;   pieces: Array<boolean>;   abort?: AbortController; } /**  * 任务/哈希值映射  */ const hashs = new Map(); /**  * 上传任务列表  */ const uploadTasks: Array<UploadTask> = []; /**  * 状态接收器  */ const statusReceivers = new Map<string, MessagePort>(); /**  * token 仓库  */ const tokenStore = {   /**    * token    */   BearerToken: '', }; /**  * 返回上传状态  * @param task 上传任务  */ const updateStatus = (task: UploadTaskInfo) => {   const taskInfo: UploadTaskInfo = {     fileName: task.fileName,     uploadPath: task.uploadPath,     id: task.id,     size: task.size,     progress: task.progress,     speed: task.speed,     status: task.status,     startTime: task.startTime,     endTime: task.endTime,     errorMessage: task.errorMessage,   };   statusReceivers.forEach((item) => {     item.postMessage(taskInfo);   }); }; /**  * 运行上传任务  * @param task 上传任务  */ const runUpload = async (task: UploadTask) => {   task.status = 'uploading';   const hash = hashs.get(task.id) || sha256.create();   hashs.set(task.id, hash);   let retryCount = 0;   const abort = new AbortController();   task.abort = abort;   while (task.status === 'uploading') {     const startTime = Date.now();     const index = task.pieces.findIndex((item) => !item);     if (index === -1) {       try {         const response: { code: number; message: string } = await fetch(           '/api/File/Upload',           {             method: 'PUT',             headers: {               Authorization: tokenStore.BearerToken,               'Content-Type': 'application/json',             },             body: JSON.stringify({               id: task.id,               fileHash: toHex(hash.digest()),               filePath: task.uploadPath,             }),           }         ).then((res) => res.json());         if (response.code !== 200) {           throw new Error(response.message);         }         task.status = 'done';         task.endTime = new Date();         updateStatus(task);       } catch (e: any) {         task.status = 'error';         task.errorMessage = e.toString();         task.endTime = new Date();         deleteUpload(task.id);         updateStatus(task);       }       break;     }     const start = index * pieceSize;     const end = start + pieceSize >= task.size ? task.size : start + pieceSize;     const buffer = task.file.slice(index * pieceSize, end);     hash.update(new Uint8Array(await buffer.arrayBuffer()));     const form = new FormData();     form.append('file', buffer);     let isTimeout = false;     try {       const timer = setTimeout(() => {         isTimeout = true;         abort.abort();       }, 8000);       const response: { code: number; message: string } = await fetch(         `/api/File/Upload?id=${task.id}&offset=${start}`,         {           method: 'POST',           body: form,           headers: {             Authorization: tokenStore.BearerToken,           },           signal: abort.signal,         }       ).then((res) => res.json());       clearTimeout(timer);       if (response.code !== 200) {         throw new Error(response.message);       }       task.pieces[index] = true;       task.progress =         task.pieces.filter((item) => item).length / task.pieces.length;       task.speed = (pieceSize / (Date.now() - startTime)) * 1000;       updateStatus(task);     } catch (e: any) {       retryCount++;       if (retryCount > 3) {         task.status = 'error';         if (isTimeout) {           task.errorMessage = 'UploadTimeout';         } else {           task.errorMessage = e.toString();         }         task.endTime = new Date();         deleteUpload(task.id);         updateStatus(task);       }     }     runNextUpload();   } }; /**  * 运行下一个上传任务  */ const runNextUpload = async () => {   if (uploadTasks.filter((item) => item.status === 'uploading').length > 3) {     return;   }   const task = uploadTasks.find((item) => item.status === 'waiting');   if (task) {     await runUpload(task);   } }; /**  * 排队上传  * @param e 消息事件  */ const queueUpload = async (   e: MessageEvent<     MessageArg<{       id: string;       file: File;       uploadPath: string;     }>   > ) => {   uploadTasks.push({     file: e.data.arg.file,     fileName: e.data.arg.file.name,     id: e.data.arg.id,     uploadPath: e.data.arg.uploadPath,     size: e.data.arg.file.size,     progress: 0,     speed: 0,     status: 'waiting',     pieces: new Array(Math.ceil(e.data.arg.file.size / pieceSize)).fill(false),     errorMessage: undefined,   });   updateStatus(uploadTasks[uploadTasks.length - 1]);   await runNextUpload(); }; /**  * 注册状态接收器  * @param e 消息事件  * @param sender 发送者  */ const registerStatusReceiver = (   e: MessageEvent<MessageArg<string>>,   sender?: MessagePort ) => {   if (sender) statusReceivers.set(e.data.arg, sender); }; /**  * 注销状态接收器  * @param e 消息事件  */ const unregisterStatusReceiver = (e: MessageEvent<MessageArg<string>>) => {   statusReceivers.delete(e.data.arg); }; /**  * 更新 token  * @param e 消息事件  */ const updateToken = (e: MessageEvent<MessageArg<string>>) => {   tokenStore.BearerToken = 'Bearer ' + e.data.arg; }; /**  * 暂停上传  * @param e 消息事件  */ const pauseUpload = (e: MessageEvent<MessageArg<string>>) => {   const task = uploadTasks.find((item) => item.id === e.data.arg);   if (task) {     task.status = 'paused';     if (task.abort) {       task.abort.abort();     }     updateStatus(task);   } }; /**  * 取消上传  * @param e 消息事件  */ const cancelUpload = (e: MessageEvent<MessageArg<string>>) => {   const task = uploadTasks.find((item) => item.id === e.data.arg);   if (task) {     task.status = 'canceled';     if (task.abort) {       task.abort.abort();     }     deleteUpload(task.id);     updateStatus(task);   } }; /**  * 删除上传  * @param id 任务 id  */ const deleteUpload = async (id: string) => {   uploadTasks.splice(     uploadTasks.findIndex((item) => item.id === id),     1   );   hashs.delete(id);   await fetch(`/api/File/Upload?id=${id}`, {     method: 'DELETE',     headers: {       Authorization: tokenStore.BearerToken,     },   }).then((res) => res.json()); }; /**  * 消息路由  */ const messageRoute = new Map<   string,   (e: MessageEvent<MessageArg<any>>, sender?: MessagePort) => void >([   ['queueUpload', queueUpload],   ['registerStatusReceiver', registerStatusReceiver],   ['updateToken', updateToken],   ['pauseUpload', pauseUpload],   ['cancelUpload', cancelUpload],   ['unregisterStatusReceiver', unregisterStatusReceiver], ]); // 监听连接 _self.onconnect = (e) => {   const port = e.ports[0];   port.onmessage = async (e) => {     // 调用函数     const func = messageRoute.get(e.data.func);     if (func) {       func(e, port);     }   };   port.start(); };  

upload-service.ts 代码

import UploadWorker from './upload-worker?sharedworker'; import { onUnmounted, ref, watch } from 'vue'; import { storeToRefs } from 'pinia'; import { useAuthStore } from 'src/stores/auth'; /**  * 上传任务信息  */ interface UploadTaskInfo {   /**    * 文件名    */   fileName: string;   /**    * 上传路径    */   uploadPath: string;   /**    * 任务 id    */   id: string;   /**    * 文件大小    */   size: number;   /**    * 上传进度    */   progress: number;   /**    * 上传速度    */   speed?: number;   /**    * 任务状态    */   status: 'uploading' | 'paused' | 'canceled' | 'done' | 'error' | 'waiting';   /**    * 开始时间    */   startTime?: Date;   /**    * 结束时间    */   endTime?: Date;   /**    * 错误信息    */   errorMessage?: string; } /**  * 上传服务  */ export const useUploadService = () => {   const store = storeToRefs(useAuthStore());   // 创建共享 worker   const worker = new UploadWorker();   /**    * 上传任务列表    */   const uploadTasks = ref<Map<string, UploadTaskInfo>>(     new Map<string, UploadTaskInfo>()   );   // 是否已注册状态接收器   const isRegistered = ref(false);   // 服务 id   const serviceId = crypto.randomUUID();   // 监听上传任务列表变化（只有在注册状态接收器后才会收到消息）   worker.port.onmessage = (e: MessageEvent<UploadTaskInfo>) => {     uploadTasks.value.set(e.data.id, e.data);   };   // 更新 token   worker.port.postMessage({     func: 'updateToken',     arg: store.token.value,   });   watch(store.token, (token) => {     worker.port.postMessage({       func: 'updateToken',       arg: token,     });   });   /**    * 排队上传    * @param file 文件    * @param uploadPath 上传路径    */   const queueUpload = (file: File, uploadPath: string) => {     worker.port.postMessage({       func: 'queueUpload',       arg: {         id: crypto.randomUUID(),         file: file,         uploadPath: uploadPath,       },     });   };   /**    * 暂停上传    * @param id 任务 id    */   const pauseUpload = (id: string) => {     worker.port.postMessage({       func: 'pauseUpload',       arg: id,     });   };   /**    * 取消上传    * @param id 任务 id    */   const cancelUpload = (id: string) => {     worker.port.postMessage({       func: 'cancelUpload',       arg: id,     });   };   /**    * 注册状态接收器    */   const registerStatusReceiver = () => {     worker.port.postMessage({       func: 'registerStatusReceiver',       arg: serviceId,     });     isRegistered.value = true;   };   /**    * 注销状态接收器    */   const unregisterStatusReceiver = () => {     worker.port.postMessage({       func: 'unregisterStatusReceiver',       arg: serviceId,     });     isRegistered.value = false;   };   onUnmounted(() => {     unregisterStatusReceiver();     worker.port.close();   });   return {     uploadTasks,     queueUpload,     pauseUpload,     cancelUpload,     registerStatusReceiver,     unregisterStatusReceiver,   }; };  

04. 用法

// 引入组合式函数 const uploadService = useUploadService(); // 注册状态接收器 uploadService.registerStatusReceiver(); // 表单绑定上传方法 const upload = (file: File, filePath: string) => {   uploadService.queueUpload(file, filePath); } // 监听上传进度，当然也可以直接展示在界面，毕竟是 Ref watch(uploadService.uploadTasks, console.log);