从 koa-body 入手分析,搞懂 Node.js 文件上传流程

柔光的暖阳◎ 2022-09-08 00:27 78阅读 0赞

# 大厂技术 高级前端 Node进阶    #

点击上方 程序员成长指北，关注公众号

回复1，加入高级Node交流群

> 作者：陈关羽 (作者投稿)
> 
> 原文地址：https://juejin.cn/post/6997060777462988837

## 故事背景 ##

某天，团队的新来的很爱问问题的小伙伴突然问我：怎么在 Koa 服务中处理接收上传的文件数据？

我答：**这个简单，你在`koa-body`里配一下`multipart`，然后在`ctx.request.files`取一下**

他又问：好的。为什么配置了`multipart`就可以在`ctx.request.files`拿到呢？

我又答：**因为`koa-body`帮你处理了**

他再问：那它是怎么处理的呢？

我：**...我去看下源码再来回答你**

## 梳理思路 ##

通过前面刨根问底的对话，我们可以提炼出三个问题：

*  WHAT：**通过什么能使 Koa 解析文件上传呢？**答：`koa-body`
 *  HOW：**如何进行配置呢？**答：开启`multipart`配置（注：更多细节在`formidable`配置中）
 *  WHY：**为什么 koa-body 可以解析上传的文件，它是什么时候在 ctx 上增加的属性呢？**

前两个问题，我们现在已经有了答案。作为一个开发者，在掌握了 WHAT 和 HOW 之后，我们更应该沉下心来，对 WHY 背后的原理进行探索。**要知其然更知其所以然**。

那么，对于这类原理性的问题，我们的思路不用多说：**看源码**

## 简单分析 koa-body 源码 ##

### 入口文件 ###

我们分析 NPM 依赖肯定要从入口文件进行分析，既**从`package.json`中的`main`字段开始，一般来说都是`index.js`**

![808f02b826ef0055429fcbca47f6ed12.png][] image.png

根据`Koa.js`的中间件实现规范，我们可以从上面的代码中了解到：

1.  `requestBody`方法返回的`function`才是是真正执行的中间件
2.  服务启动时，`requestBody`方法会初始化相关配置

### requestBody 返回中间件实现 ###

![89fe145f333000a8095b2f07db3d6828.png][] image.png

接下来我们来看中间件真正实现的代码。

由于屏幕原因我对一些无关的代码进行了折叠，我们重点看红框处。

1.  当`opts.multipart`（配置验证）与`ctx.is('multipart')`（请求头验证）都为`true`时，判断为文件上传场景，调用`formy`方法
2.  当`formy`方法返回的`promise`实例`resolved`时，将`promise`实例返回的数据附加到`ctx.request.body`及`ctx.request.files`上

这下 WHY 的后半部分谜底解开了：**在真实处理逻辑的`promise`实例 resolved 后，`koa-body`会将返回的数据附加在`ctx.request`上**

### formy 方法实现 ###

通过上一部分的截图，我们可以看到对文件解析的逻辑都在`formy`方法中。那么这个方法是怎么实现的呢。

话不多说，上代码

function formy(ctx, opts) {
      return new Promise(function (resolve, reject) {
        var fields = {};
        var files = {};
        var form = new forms.IncomingForm(opts);
        form.on('end', function () {
          return resolve({
            fields: fields,
            files: files
          });
        }).on('error', function (err) {
          return reject(err);
        }).on('field', function (field, value) {
          if (fields[field]) {
            if (Array.isArray(fields[field])) {
              fields[field].push(value);
            } else {
              fields[field] = [fields[field], value];
            }
          } else {
            fields[field] = value;
          }
        }).on('file', function (field, file) {
          if (files[field]) {
            if (Array.isArray(files[field])) {
              files[field].push(file);
            } else {
              files[field] = [files[field], file];
            }
          } else {
            files[field] = file;
          }
        });
        if (opts.onFileBegin) {
          form.on('fileBegin', opts.onFileBegin);
        }
        form.parse(ctx.req);
      });
    }

逻辑不多，我们再来总结一下。

1.  new 了一个 `IncomingForm`实例，传入`formidable`对应的配置
2.  调用实例的`parse`方法，并监听`end`、`error`、`file`、`field`等事件
3.  在`end`事件的回调中，进行`resolve`

那么`forms.IncomingForm`是哪来的呢？

const forms = require('formidable');

原来是`koa-body`引用的**第三方依赖`formidable`**

这下我们明确了，**`ctx.request`对象上附加的数据是在`formidable.IncomingForm`实例中进行处理，通过`file`、`field`等事件回调进行接收，最后在`end`事件回调中进行返回的**。

## 简单分析 formidable 源码 ##

### 入口文件 ###

通过前面的分析，我们知道了 `koa-body` 对于文件的处理是引用的 `formidable`。我们还是从入口文件进行分析。

![cb822b5ea3ffce6f1aee1b408a998c8f.png][] image.png

入口代码非常简单，核心逻辑看来都在`Formidable`中

### Formidable.js 分析 ###

![be67c843e4a7a729f8fa0434f07b18df.png][] image.png

先来对`Formidable.js`有一个宏观印象：

1.  定义并导出了 `IncomingForm` 类
2.  `IncomingForm`继承了`EventEmitter`
3.  `IncomingForm`定义了诸多方法

**通过继承`EventEmitter`，`IncomingForm`实例可以在合适的时机触发相关事件并监听处理**

### parse 过程 ###

回到上文，`IncomingForm`实例调用了`parse`方法。我们从`parse`方法开始入手分析。

![4c40dc6a64602b21bc2f0fa7008cd7b4.png][] image.png

通过红框处逻辑我们可以看到，`parse`方法的职责主要有两个：

1.  **解析请求 header**，设置 parser
2.  **监听`req`参数的`data`事件**，处理数据。

通过前面传入的参数，我们知道 **`req`参数就是`ctx.req`，`Node.js`原生`request`对象**。

这下谜底又解开了一部分，**`koa-body`是怎么拿到上传的文件数据的呢？通过监听`Node.js`原生`request`对象的`data`事件**

### write 过程 ###

本节涉及到很多方法嵌套调用，我统称为`write`过程。

`writeHeaders`方法的调用链是这样的（代码比较长，大家感兴趣的可以自行去看源码）：

`writeHeaders()` \-> `_parseContentType()` \-> `initMultipart()` \-> `MultipartParser`

经过一串复杂的调用，我们完成了`writeHeaders`方法的使命：**解析请求 header，设置 parser**

#### MultipartParser 源码分析 ####

![ff1eed1cc1432ce10e6483628fedc477.png][] image.png

我们可以看到，**定义的`MultipartParser`类继承了`Transform`流**

我们简单看下官方文档对于`Tranform`流的定义

![dd03e91532547441fd7aba549b8f1f90.png][] image.png

通过定义的`_transform`方法，我们可以对写进流的`buffer`进行处理

_transform(buffer, _, done) {
        let i = 0;
        let prevIndex = this.index;
        let { index, state, flags } = this;
        const { lookbehind, boundary, boundaryChars } = this;
        const boundaryLength = boundary.length;
        const boundaryEnd = boundaryLength - 1;
        this.bufferLength = buffer.length;
        let c = null;
        let cl = null;
    
        const setMark = (name, idx) => {
          this[`${name}Mark`] = typeof idx === 'number' ? idx : i;
        };
    
        const clearMarkSymbol = (name) => {
          delete this[`${name}Mark`];
        };
    
        const dataCallback = (name, shouldClear) => {
          const markSymbol = `${name}Mark`;
          if (!(markSymbol in this)) {
            return;
          }
    
          if (!shouldClear) {
            this._handleCallback(name, buffer, this[markSymbol], buffer.length);
            setMark(name, 0);
          } else {
            this._handleCallback(name, buffer, this[markSymbol], i);
            clearMarkSymbol(name);
          }
        };
    
        for (i = 0; i < this.bufferLength; i++) {
          c = buffer[i];
          switch (state) {
            case STATE.PARSER_UNINITIALIZED:
              return i;
            case STATE.START:
              index = 0;
              state = STATE.START_BOUNDARY;
            case STATE.START_BOUNDARY:
              if (index === boundary.length - 2) {
                if (c === HYPHEN) {
                  flags |= FBOUNDARY.LAST_BOUNDARY;
                } else if (c !== CR) {
                  return i;
                }
                index++;
                break;
              } else if (index - 1 === boundary.length - 2) {
                if (flags & FBOUNDARY.LAST_BOUNDARY && c === HYPHEN) {
                  this._handleCallback('end');
                  state = STATE.END;
                  flags = 0;
                } else if (!(flags & FBOUNDARY.LAST_BOUNDARY) && c === LF) {
                  index = 0;
                  this._handleCallback('partBegin');
                  state = STATE.HEADER_FIELD_START;
                } else {
                  return i;
                }
                break;
              }
    
              if (c !== boundary[index + 2]) {
                index = -2;
              }
              if (c === boundary[index + 2]) {
                index++;
              }
              break;
            case STATE.HEADER_FIELD_START:
              state = STATE.HEADER_FIELD;
              setMark('headerField');
              index = 0;
            case STATE.HEADER_FIELD:
              if (c === CR) {
                clearMarkSymbol('headerField');
                state = STATE.HEADERS_ALMOST_DONE;
                break;
              }
    
              index++;
              if (c === HYPHEN) {
                break;
              }
    
              if (c === COLON) {
                if (index === 1) {
                  // empty header field
                  return i;
                }
                dataCallback('headerField', true);
                state = STATE.HEADER_VALUE_START;
                break;
              }
    
              cl = lower(c);
              if (cl < A || cl > Z) {
                return i;
              }
              break;
            case STATE.HEADER_VALUE_START:
              if (c === SPACE) {
                break;
              }
    
              setMark('headerValue');
              state = STATE.HEADER_VALUE;
            case STATE.HEADER_VALUE:
              if (c === CR) {
                dataCallback('headerValue', true);
                this._handleCallback('headerEnd');
                state = STATE.HEADER_VALUE_ALMOST_DONE;
              }
              break;
            case STATE.HEADER_VALUE_ALMOST_DONE:
              if (c !== LF) {
                return i;
              }
              state = STATE.HEADER_FIELD_START;
              break;
            case STATE.HEADERS_ALMOST_DONE:
              if (c !== LF) {
                return i;
              }
    
              this._handleCallback('headersEnd');
              state = STATE.PART_DATA_START;
              break;
            case STATE.PART_DATA_START:
              state = STATE.PART_DATA;
              setMark('partData');
            case STATE.PART_DATA:
              prevIndex = index;
    
              if (index === 0) {
                // boyer-moore derrived algorithm to safely skip non-boundary data
                i += boundaryEnd;
                while (i < this.bufferLength && !(buffer[i] in boundaryChars)) {
                  i += boundaryLength;
                }
                i -= boundaryEnd;
                c = buffer[i];
              }
    
              if (index < boundary.length) {
                if (boundary[index] === c) {
                  if (index === 0) {
                    dataCallback('partData', true);
                  }
                  index++;
                } else {
                  index = 0;
                }
              } else if (index === boundary.length) {
                index++;
                if (c === CR) {
                  // CR = part boundary
                  flags |= FBOUNDARY.PART_BOUNDARY;
                } else if (c === HYPHEN) {
                  // HYPHEN = end boundary
                  flags |= FBOUNDARY.LAST_BOUNDARY;
                } else {
                  index = 0;
                }
              } else if (index - 1 === boundary.length) {
                if (flags & FBOUNDARY.PART_BOUNDARY) {
                  index = 0;
                  if (c === LF) {
                    // unset the PART_BOUNDARY flag
                    flags &= ~FBOUNDARY.PART_BOUNDARY;
                    this._handleCallback('partEnd');
                    this._handleCallback('partBegin');
                    state = STATE.HEADER_FIELD_START;
                    break;
                  }
                } else if (flags & FBOUNDARY.LAST_BOUNDARY) {
                  if (c === HYPHEN) {
                    this._handleCallback('partEnd');
                    this._handleCallback('end');
                    state = STATE.END;
                    flags = 0;
                  } else {
                    index = 0;
                  }
                } else {
                  index = 0;
                }
              }
    
              if (index > 0) {
                // when matching a possible boundary, keep a lookbehind reference
                // in case it turns out to be a false lead
                lookbehind[index - 1] = c;
              } else if (prevIndex > 0) {
                // if our boundary turned out to be rubbish, the captured lookbehind
                // belongs to partData
                this._handleCallback('partData', lookbehind, 0, prevIndex);
                prevIndex = 0;
                setMark('partData');
    
                // reconsider the current character even so it interrupted the sequence
                // it could be the beginning of a new sequence
                i--;
              }
    
              break;
            case STATE.END:
              break;
            default:
              return i;
          }
        }
    
        dataCallback('headerField');
        dataCallback('headerValue');
        dataCallback('partData');
    
        this.index = index;
        this.state = state;
        this.flags = flags;
    
        done();
        return this.bufferLength;
      }

`_transform`的逻辑也很简单：**遍历`buffer`，在不同的阶段进行不同的处理并通过`_handleCallback`方法发出不同的事件进而触发对应的回调**。

注：`_handleCallback`方法实现很有意思，有兴趣的同学可以自己看下

#### parser.on('data') ####

上文我们说到，**当`parser`流被写入数据时，会调用`_transform`方法进行处理，同时触发各个周期事件的回调**。

事件回调的代码如图所示：

![22a7b5dba8887aa8dc13f41d285ca822.png][] image.png

我们需要关注的是`headersEnd`事件，**在`headsEnd`事件的回调中，调用了`IncomingForm`实例的`onPart`方法**

**为什么说`this.onPart`调用的是`IncomingForm`实例的方法呢，可以看下前面的代码，有一步`call`绑定`this`的操作**

在层层嵌套调用中，我们终于回到了`IncomingForm`的逻辑里，可喜可贺。

#### write 过程简单总结 ####

我们再来回顾一下`write`过程：

*  `this.writeHeaders(req.headers);` \- 设置`parser`
 *  `req.on('data')` \- 调用`parser`的`write`方法，向流中写入数据
 *  `parser.on('headersEnd')` \- `headers`解析完毕，调用`this.onPart`方法

### 保存过程 ###

#### `_handlePart` 方法源码分析 ####

当`headers`解析完毕之后，我们会调用`this.onPart`方法。而`this.onPart`方法的核心逻辑都在`_handlePart`方法中。接下来我们来对`_handlePart`源码进行分析

_handlePart(part) {
        if (part.originalFilename && typeof part.originalFilename !== 'string') {
          this._error(
            new FormidableError(
              `the part.originalFilename should be string when it exists`,
              errors.filenameNotString,
            ),
          );
          return;
        }
    
        if (!part.mimetype) {
          let value = '';
          const decoder = new StringDecoder(
            part.transferEncoding || this.options.encoding,
          );
    
          part.on('data', (buffer) => {
            this._fieldsSize += buffer.length;
            if (this._fieldsSize > this.options.maxFieldsSize) {
              this._error(
                new FormidableError(
                  `options.maxFieldsSize (${this.options.maxFieldsSize} bytes) exceeded, received ${this._fieldsSize} bytes of field data`,
                  errors.maxFieldsSizeExceeded,
                  413, // Payload Too Large
                ),
              );
              return;
            }
            value += decoder.write(buffer);
          });
    
          part.on('end', () => {
            this.emit('field', part.name, value);
          });
          return;
        }
    
        if (!this.options.filter(part)) {
          return;
        }
    
        this._flushing += 1;
    
        const newFilename = this._getNewName(part);
        const filepath = this._joinDirectoryName(newFilename);
    
        // new 了一个 File 实例
        const file = this._newFile({
          newFilename,
          filepath,
          originalFilename: part.originalFilename,
          mimetype: part.mimetype,
        });
        file.on('error', (err) => {
          this._error(err);
        });
        this.emit('fileBegin', part.name, file);
    
        // 调用 open 方法
        file.open();
        this.openedFiles.push(file);
    
        // 监听 data 时间
        part.on('data', (buffer) => {
          this._fileSize += buffer.length;
          if (this._fileSize < this.options.minFileSize) {
            this._error(
              new FormidableError(
                `options.minFileSize (${this.options.minFileSize} bytes) inferior, received ${this._fileSize} bytes of file data`,
                errors.smallerThanMinFileSize,
                400,
              ),
            );
            return;
          }
          if (this._fileSize > this.options.maxFileSize) {
            this._error(
              new FormidableError(
                `options.maxFileSize (${this.options.maxFileSize} bytes) exceeded, received ${this._fileSize} bytes of file data`,
                errors.biggerThanMaxFileSize,
                413,
              ),
            );
            return;
          }
          if (buffer.length === 0) {
            return;
          }
          this.pause();
    
          // 写入 file
          file.write(buffer, () => {
            this.resume();
          });
        });
    
        // 监听 end 事件
        part.on('end', () => {
          if (!this.options.allowEmptyFiles && this._fileSize === 0) {
            this._error(
              new FormidableError(
                `options.allowEmptyFiles is false, file size should be greather than 0`,
                errors.noEmptyFiles,
                400,
              ),
            );
            return;
          }
    
          // 关闭事件
          file.end(() => {
            this._flushing -= 1;
            this.emit('file', part.name, file);
            this._maybeEnd();
          });
        });
      }

通过我在代码关键处加的注释，我们可以看到`_handlePart`方法的核心逻辑如下：

1.  `new` 一个 `File` 对象的实例 `file`
2.  开启 `file`
3.  监听 `part` 的 `data` 事件，将数据写入 `file` 中
4.  监听 `part` 的 `end` 事件，关闭 `file`

在解读这段核心逻辑前，我们需要明确**两个重要信息**：

1.  传入的参数 `part` 是什么？
    
    **`part`是在`initMultipart`方法中创建的可读流，通过`data`事件向外界传输数据**

![88f0537ffb28a266b42d0e0f7a0b4361.png][] image.png

1.  `File` 对象的实例 `file` 是什么？
    
    **是根据传入的`filePath`创建的可写流**

![ad9a1e0e65528ccb5d0615e1ac109669.png][] image.png

明白了这两个前提，这下我们终于明白了！

**`_handlePart`方法就是打开一个文件流，把`parser`解析出的数据通过文件流写入文件，然后关闭**。

### 结束流程 ###

#### \_maybeEnd ####

我们的分析终于来到了尾声！

**整体流程如何结束呢？**

我们注意到有一个方法叫`_maybeEnd`，**当满足条件时，会触发`end`事件**

![89b812513de85e9c0e52c072f46e1824.png][] image.png

我们选取`this._flushing`变量进行分析，如何满足条件。

可以看到，`this._flushing`初始值为 0，当`_handlePart`方法执行时，`this._flushing+=1`；当`file`关闭时，`this._flushing-=1`。

由此我们可以分析出，**当上传的文件都处理完毕时，`this._flushing`变量恢复为 0，满足条件，触发`end`事件**

大家看到`end`事件应该很兴奋，因为我们终于走到结束了。

#### 真正结束 ####

`end`事件被谁接收，相信大家心里已经有数了。

没错，我们回到了`koa-body`的代码中。后面的流程如下：

1.  **在`formy`方法返回的`promise`实例中监听到`IncomingForm`实例发出的`end`事件，`promise`实例进行`resolve`**![b8a8af60f73180b47156c03d6e1764b2.png][]
2.  **在`promise`实例的`then`中接收到`resolve`返回的数据，附加到`ctx.request`对象上**![8b1681e7d7f051d6b2d20284f7f9c4d5.png][]
3.  **中间件执行结束，调用`next()`**

## 回答问题 ##

本文带领大家从一个文件上传的例子入手，分析了`koa-body`及`formidable`中关于处理文件上传的核心逻辑。

对于我们前面遗留的问题，相信大家此时已经有了答案。

简单回答，koa-body 是如何处理文件上传的呢？

答：

1.  通过`req.on('data')`获取数据
2.  解析`header`，解析`boundary`
3.  通过文件流写入本地文件中

## 分析总结 ##

除了对`koa-body`文件上传流程有了清晰的了解之外，在整体探索分析过程中，我们还应该有一些别的收获，比如

*  遇事不明，**读源码**
 *  对于一些依赖，**源码与`node_modules`中下载的代码格式不同，二者对比阅读有奇效**
 *  我们应该有**使用流操作文件**的意识
 *  **`EventEmitter`是通信神器，这种思路可以利用到业务代码中**

Node 社群

我组建了一个氛围特别好的 Node.js 社群，里面有很多 Node.js小伙伴，如果你对Node.js学习感兴趣的话（后续有计划也可以），我们可以一起进行Node.js相关的交流、学习、共建。下方加 考拉 好友回复「Node」即可。

![178a94cea1813c0fa47dfaa02deb9676.png][]

**“分享、点赞、在看” 支持一波** ![1aad187a2759b6eae11b16aa79149f29.png][]

[808f02b826ef0055429fcbca47f6ed12.png]: /images/20220829/9f03da15dd514723bfd66f70436ab366.png
[89fe145f333000a8095b2f07db3d6828.png]: /images/20220829/8b1281031e904f819f7b5eb294180c2e.png
[cb822b5ea3ffce6f1aee1b408a998c8f.png]: /images/20220829/f9df314f0f6a48e488606632e16eb7cb.png
[be67c843e4a7a729f8fa0434f07b18df.png]: /images/20220829/7d82e57798144b60aabd6f4f677cb636.png
[4c40dc6a64602b21bc2f0fa7008cd7b4.png]: /images/20220829/1c9c27a0aa3b44609601d49237d28fcd.png
[ff1eed1cc1432ce10e6483628fedc477.png]: /images/20220829/9d0a42a069cc4fdf8639f9461b5c866f.png
[dd03e91532547441fd7aba549b8f1f90.png]: /images/20220829/f9694b869611489ca389daefbf826b77.png
[22a7b5dba8887aa8dc13f41d285ca822.png]: /images/20220829/725e74d80b27484f837ba92c7faed473.png
[88f0537ffb28a266b42d0e0f7a0b4361.png]: /images/20220829/2cedf9fc5353449f992967454dd37871.png
[ad9a1e0e65528ccb5d0615e1ac109669.png]: /images/20220829/70f4f8faa382476babdeef0db92d6013.png
[89b812513de85e9c0e52c072f46e1824.png]: /images/20220829/3e90a0ac75e546419ab4b3d0e73d0b3c.png
[b8a8af60f73180b47156c03d6e1764b2.png]: /images/20220829/7c7bdd8157d0412bbba24f519bb7ad1b.png
[8b1681e7d7f051d6b2d20284f7f9c4d5.png]: /images/20220829/631aee9dd078411f80d8a930e67edfce.png
[178a94cea1813c0fa47dfaa02deb9676.png]: /images/20220829/417e9e579a6745679b2cf00a86da5d3e.png
[1aad187a2759b6eae11b16aa79149f29.png]: /images/20220829/5290c95917b345418d5c274e7504ebad.png