斌の顶级理解

Improve a little bit every day.

0%

Web Worker

发表于 更新于 分类于

Web Worker是Web浏览器提供的一种在后台线程中运行JavaScript的功能。它独立于主线程运行，可以执行计算密集型或长时间运行的任务，而不会阻塞页面的渲染和交互。通过将大文件切片上传的逻辑放在Web Worker中执行，我们可以充分利用浏览器的多线程能力，提高上传速度，并保持页面的流畅运行。

基于Vue的基础用法

在Vue项目中配置webpack来使用web-worker涉及几个关键步骤。这主要涉及到处理worker文件的加载，确保它们被正确地打包和引用。以下是一个基本的配置过程：

1.安装worker-loader

npm install --save-dev worker-loader

2.配置webpack
1
2
3
4
5
6
7
8
9
10
11
12
13
module.exports = {
  publicPath: './',

  chainWebpack: config => {  
    config.module  
       .rule('worker')  
      .test(/\.worker\.js$/)  // 如果需要.worker.js后缀
      .use('worker-loader')  
      .loader('worker-loader')
      .options({ // 可以查阅worker-loader文档，根据自己的需求进行配置
       })
  }  
}
3.创建和使用worker

创建一个worker文件，并给它一个.worker.js的扩展名。例如，你可以创建一个my-worker.worker.js文件。

1
2
3
4
5
6
7
8
9
10
11
// my-worker.worker.js  
self.onmessage = function(e) {  
  console.log('Worker: Hello World');  
  const result = doSomeWork(e.data);  
  self.postMessage(result);  
};  
  
function doSomeWork(data) {  
  // 模拟一些工作  
  return data * 2;  
}

在你的Vue组件或其他JavaScript文件中，你可以像下面这样创建一个worker实例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
// MyComponent.vue 或其他.js文件  
import MyWorker from './my-worker.worker.js';  
  
export default {  
  methods: {  
    startWorker() {  
      const myWorker = new MyWorker();  
  
      myWorker.onmessage = (e) => {  
        console.log('Main script: Received result', e.data);  
      };  
  
      myWorker.postMessage(100); // 发送数据给worker  
    }  
  },  
  mounted() {  
    this.startWorker();  
  }  
};

实战：实现大文件切片上传

1.逻辑梳理

文件切片：使用 JavaScript 的 Blob.prototype.slice() 方法将大文件切分成多个切片。

上传切片：使用 axios 或其他 HTTP 客户端库逐个上传切片。可以为每个切片生成一个唯一的标识符（例如，使用文件的哈希值和切片索引），以便后端能够正确地将它们合并。

客户端线程数：获取用户CPU线程数量，以便最大优化上传文件速度。

控制上传接口的并发数量：防止大量的请求并发导致页面卡死，设计一个线程队列，控制请求数量一直保持在6。

2.实现
1.获取客户端线程数量

navigator.hardwareConcurrency 是一个只读属性，它返回用户设备的逻辑处理器内核数。

1
export const getConcurrency = () => navigator.hardwareConcurrency || 4 // 浏览器不支持就默认4核
2.主线程

定义和处理一些必要的常量，并且根据用户的线程数进行开启多线程Web-worker任务处理文件切片。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
import { defer, createEventHandler } from 'js-hodgepodge'
import FileWorker from './files.worker'

export const getConcurrency = () => navigator.hardwareConcurrency || 4

export const handleEvent = () => createEventHandler('handleSchedule')

export const sliceFile = file => {

  const dfd = defer()
  
  const chunkSize = 1024 // 1Kb
  const thread = getConcurrency() // 线程数

  const chunks = []
  const chunkNum = Math.ceil(file.size / chunkSize) // 切片总数量

  const workerChunkCount = Math.ceil(chunkNum / thread) // 每个线程需要处理的切片数量
  let finishCount = 0;

  for (let i = 0; i < thread; i++) {

    const worker = new FileWorker()

    // 计算每个线程的开始索引和结束索引
    const startIndex = i * workerChunkCount;

    let endIndex = startIndex + workerChunkCount;

    // 防止最后一个线程结束索引大于文件的切片数量的总数量
    if (endIndex > chunkNum) {
      endIndex = chunkNum;
    }

    worker.postMessage({
      file,
      chunkSize,
      startIndex,
      endIndex,
    });

    worker.onmessage = (e) => {

      // 接收到 worker 线程返回的消息
      for (let i = startIndex; i < endIndex; i++) {

        chunks[i] = {
          ...e.data[i - startIndex],
          chunkNum,
          filename: file.name
        };

      }

      worker.terminate(); // 关闭线程

      finishCount++;

      if (finishCount === thread) {
        
        dfd.resolve({
          chunks,
          chunkNum
        });
      }
    };

  }

  return dfd
}
3.实现文件切片

首先，我们需要创建一个 Web Worker 脚本，用于处理文件切片和切片hash

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
import md5 from 'js-md5'

self.onmessage = async function ({
  data: {
    file,
    chunkSize,
    startIndex,
    endIndex,
  }
}) {

  const arr = [];

  for (let i = startIndex; i < endIndex; i++) {
    arr.push(
      createChunks(file, i, chunkSize)
    );
  }
  const chunks = await Promise.all(arr)

  // 提交线程信息
  postMessage(chunks);
}

const createChunks = (
  file,
  index,
  chunkSize
) => {
  return new Promise((resolve) => {

    // 开始第几个*分片的大小
    const start = index * chunkSize;

    // 结束时start + 分片的大小
    const end = start + chunkSize;
    const fileReader = new FileReader();

    // 每个切片都通过FileReader读取为ArrayBuffer
    fileReader.onload = (e) => {

      const content = new Uint8Array(e.target.result);
      const files = file.slice(start, end);

      const md5s = md5.arrayBuffer(content)

      function arrayBufferToHex(buffer) {
        let bytes = new Uint8Array(buffer);
        let hexString = '';
        for (let i = 0; i < bytes.byteLength; i++) {
          let hex = bytes[i].toString(16);

          hexString += hex.length === 1 ? '0' + hex : hex;
        }
        return hexString;
      }

      resolve({
        start,
        end,
        index,
        hash: arrayBufferToHex(md5s),  // 生成唯一的hash
        files,
      });
    };

    // 读取文件的分片
    fileReader.readAsArrayBuffer(file.slice(start, end));
  });
}

Web Worker通过onmessage事件接收消息。当主线程发送消息时，这个消息会作为参数传递给onmessage函数。

切片hash处理流程：使用FileReader来读取文件内容。当文件分片读取完毕后，会触发onload这个事件,使用new Uint8Array(e.target.result)将读取的ArrayBuffer转换为Uint8Array，再利用js-md5的使用md5.arrayBuffer(content)计算分片的MD5哈希值，使用arrayBufferToHex函数将切片buffer转换为十六进制String，当所有分片处理完毕后，将结果（即分片及其相关信息）发送postMessage回主线程。

4.请求池的设计与处理

我这里创建一个请求队列，并使用 Promise 来控制并发请求的数量。创建一个数组来存储待处理的请求，并使用 Promise 来控制每次只有一定数量的请求被发送。当某个请求完成时，再从队列中取出下一个请求来发送。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
export const uploadFile = (
  chunks // 总切片
) => {
  chunks = chunks || []

  let schedule = 0 // 进度

  const { dispatch } = handleEvent()

  const requestQueue = (concurrency) => {
    concurrency = concurrency || 6
    const queue = [] // 线程池
    let current = 0

    const dequeue = () => {
      while (current < concurrency && queue.length) {
        current++;
        const requestPromiseFactory = queue.shift();
        requestPromiseFactory()
          .then(result => { // 上传成功处理
            console.log(result)

            schedule++; // 收集上传切片成功的数量

            dispatch(window, schedule);  // 事件派发，通知进度
          })
          .catch(error => { // 失败
            console.log(error)
          })
          .finally(() => {
            current--;
            dequeue();
          });
      }

    }

    return (requestPromiseFactory) => {
      queue.push(requestPromiseFactory)
      dequeue()
    }

  }

  const handleFormData = obj => {
    const formData = new FormData()

    Object
      .entries(obj)
      .forEach(([key, val]) => {
        formData.append(key, val)
      })

    return formData
  }

  const enqueue = requestQueue(6)

  for (let i = 0; i < chunks.length; i++) {

    enqueue(() => axios.post(
      '/api/upload',
      handleFormData(chunks[i]),
      {
        headers: {
          'Content-Type': 'multipart/form-data' 
        }
      }
    ))
  }

  return schedule

}

利用了第三方库js-hodgepodge的发布订阅，将上传切片成功的数量发布给主界面，得到相应的上传进度。

5.主界面代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
<template>
  <div>
    <input type="file" ref="file">

    <button @click="handleUpload">提交</button>

    <p>进度：{{ progress * 100 }}%</p>
  </div>
</template>

<script>
import { sliceFile, uploadFile, handleEvent } from './file.utils'
export default {

  data() {
    return {
      progress: 0
    }
  },

  methods: {
    async handleUpload() {
      const file = this.$refs.file.files[0]
    
      if(!file) {
        return
      }

      console.time()

      const dfd = sliceFile(file)

      dfd
        .promise
        .then(({ chunks, chunkNum }) => {
          uploadFile(chunks)

          const { addEventListener } = handleEvent()

          const eject = addEventListener(window, ({ detail: schedule }) => {

            this.progress = schedule / chunkNum

            if(schedule === chunkNum) { // 上传完成，关闭事件监听
              eject()
            }
          })
        })

      console.timeEnd() 
    }
  }
}
</script>

<style>

</style>
6.执行响应结果打印

当执行一个大文件上传时，时间可被大大的压缩了。

640

node后端切片与组合结果微信图片_20241113171134

其实整个流程比较重要的就是文件切片，和请求池的设计

新一代并发方案

SharedArrayBuffer的威力
1
2
3
4
5
6
7
8
// 主线程
const sharedBuffer =new SharedArrayBuffer(1024)
const arr =new Int32Array(sharedBuffer);
// Worker线程
self.onmessage =function(e){
  const sharedBuffer = e.data;
  const arr =new Int32Array(sharedBuffer);// 多线程可同时操作
}
WebAssembly + Worker
1
2
3
4
5
6
const worker = new Worker('wasm-worker.js');
worker.postMessage({wasmFile:'compute-heavy.wasm'});

worker.onmessage=(e)=>{
	console.l0g('WASM计算结果:'，e.data);
}
-------------本文结束感谢您的阅读-------------