eggjs

2024年01月29日约1.4k字预计需要6分钟

migration

npx sequelize init:config
npx sequelize init:migrations
# 首先编写migration初始化表
npx sequelize migration:generate --name=init-users
# 升级数据库
npx sequelize db:migrate
npx sequelize db:migrate --env development
# 如果有问题需要回滚，可以通过 `db:migrate:undo` 回退一个变更
# npx sequelize db:migrate:undo
# 可以通过 `db:migrate:undo:all` 回退到初始状态
# npx sequelize db:migrate:undo:all

文件上传

file模式

1
2
3

exports.multipart = {
  mode: 'file',
};

- ctx.request.body: Get all the multipart fields and values, except file.

- ctx.request.files: Contains all file from the multipart request, it’s an Array object.

stream模式

config:

config.multipart = {
    fileSize: '5000mb',
    mode: 'stream',
    tmpdir: path.join(process.cwd(), 'paks_temp'),
    whitelist: (filename) => {
        let ext = filename.substr(filename.lastIndexOf('.'));
        console.log(`multipart_upload filename: ${filename} ext:${ext}`);
        if(ext == '.exe') return false;
        return true;
    }
}

单文件

// 前端
function uploadFile(file, { xx1, xx2, xx3 }) {
  if (!(file instanceof File)) {
    throw new Error("无效的文件对象");
  }
  const forms = new FormData();
  // 参数
  forms.append("xx1", xx1);
  forms.append("xx2", xx2);
  forms.append("xx3", xx3);
  // 文件 文件放最后
  forms.append("file", file);
  const url ="<https://example.com/upload>";
  return axios.post(url, forms);
}

nodejs

const awaitWriteStream = require('await-stream-ready').write;
async upload() {
    let { ctx } = this;
    let stream;
    let streamErrCode;
    try {
      stream = await this.ctx.getFileStream({ limits: {
        fileSize: 3 * 1024 * 1024
      }});
      // on limit
      stream.on('error', (error: any) => {
        streamErrCode = error.status;
        ctx.logger.error('[Controller][upload] getFileStream on error:', error);
      });
    } catch (error: any) {
      // truncated
      streamErrCode = error.status;
      ctx.logger.error('[Controller][upload] getFileStream error:', error);
    }
    const uplaodBasePath = this.config.multipart.tmpdir;
    let filename = stream.filename;
    const source = path.join(uplaodBasePath, `${filename}.${new Date().getTime()} `);// 源路径
      const writeStream = fs.createWriteStream(source);
      await awaitWriteStream(stream.pipe(writeStream));
      if (streamErrCode === 413) {
        return ctx.body = new ResultJSON().fail(ApiCode.virtual_img_size_limit, '图片大小超出限制');
      }
      let params = stream.fields;
      let data = JSON.parse(params.data || {});
      let { gid, imgId } = data;
      const dir = `${process.cwd()}/paks/`;
    // 生成写入路径
      let imgName = `${new Date().getTime()}${filename.substr(filename.lastIndexOf("."))}`;
   	  let imageDir = `/ai_virtual_bg/${imgDB.gid}`;
      let dbPath =  path.join(imageDir, imgName);
      let target = path.join(dir, dbPath); // 目标路径存储
      fs.copyFileSync(source, target);
      this.ctx.body = new ResultJSON().success();
  }

多文件

const sendToWormhole = require('stream-wormhole');
const Controller = require('egg').Controller;

class UploaderController extends Controller {
  async upload() {
    const ctx = this.ctx;
    const parts = ctx.multipart();
    let part;
    // parts() 返回 promise 对象
    while ((part = await parts()) != null) {
      if (part.length) {
        // 这是 busboy 的字段
        console.log('field：' + part[0]);
        console.log('value：' + part[1]);
        console.log('valueTruncated：' + part[2]);
        console.log('fieldnameTruncated：' + part[3]);
      } else {
        if (!part.filename) {
          // 这时是用户没有选择文件就点击了上传（part 是 file stream，但是 part.filename 为空）
          // 需要做出处理，例如给出错误提示消息
          return;
        }
        // part 是上传的文件流
        console.log('field：' + part.fieldname);
        console.log('filename：' + part.filename);
        console.log('encoding：' + part.encoding);
        console.log('mime：' + part.mime);
        // 文件处理，上传到云存储等等
        let result;
        try {
          result = await ctx.oss.put('egg-multipart-test/' + part.filename, part);
        } catch (err) {
          // 必须将上传的文件流消费掉，要不然浏览器响应会卡死
          await sendToWormhole(part);
          throw err;
        }
        console.log(result);
      }
    }
    console.log('and we are done parsing the form!');
  }
}

module.exports = UploaderController;

for-await-of

for await of stream会消费流吗

是的，当使用 for-await-of 循环迭代可异步迭代对象时，每次迭代都会消耗一部分流。

在异步迭代过程中，for-await-of 循环会暂停并等待流产生下一个可用的数据项（如 chunk 或值）。然后，循环将处理该数据项，并继续等待下一个数据项。

每次迭代都会消耗一个数据项，直到遍历结束或流被消耗完毕。当流被完全消耗时，迭代也将终止。

这种行为类似于通过调用 stream.read() 或使用可读流的其他读取机制逐步消耗流的数据。每次读取操作都会从流中消耗一部分数据。

如果您需要在使用 for-await-of 循环迭代流后继续操作流的剩余数据，您需要根据需要对其进行处理，例如使用转换流或其他流处理技术将流数据保存起来，以便后续操作。

1 2	`npm search stream clone`

定时任务

interval

const Subscription = require('egg').Subscription;

class UpdateCache extends Subscription {
  // 通过 schedule 属性来设置定时任务的执行间隔等配置
  static get schedule() {
    return {
      interval: '1m', // 1 分钟间隔
      type: 'all', // 指定所有的 worker 都需要执行
      // cron: '0 0 */3 * * *',
    };
  }

  // subscribe 是真正定时任务执行时被运行的函数
  async subscribe() {
    const res = await this.ctx.curl('http://www.api.com/cache', {
      dataType: 'json',
    });
    this.ctx.app.cache = res.data;
  }
}

module.exports = UpdateCache;

框架提供的定时任务默认支持两种类型，worker 和 all。worker 和 all 都支持上面的两种定时方式，只是当到执行时机时，会执行定时任务的 worker 不同：

worker 类型：每台机器上只有一个 worker 会执行这个定时任务，每次执行定时任务的 worker 的选择是随机的。
all 类型：每台机器上的每个 worker 都会执行这个定时任务。

cron

注意：cron-parser 支持可选的秒（linux crontab 不支持）。

* * * * * *
┬ ┬ ┬ ┬ ┬ ┬
│ │ │ │ │ │
│ │ │ │ │ └─ 周日（0 - 7）（0 或 7 是周日）
│ │ │ │ └─── 月份（1 - 12）
│ │ │ └───── 日期（1 - 31）
│ │ └─────── 小时（0 - 23）
│ └───────── 分钟（0 - 59）
└─────────── 秒（0 - 59，可选）

例如：

1 2	`0 30 12 * * * // 每天12:30 0 30 12 * * 3 // 每周三12:30`

动态配置

module.exports = (app) => {
  return {
    schedule: {
      interval: app.config.cacheTick,
      type: 'all',
    },
    async task(ctx) {
      const res = await ctx.curl('http://www.api.com/cache', {
        contentType: 'json',
      });
      ctx.app.cache = res.data;
    },
  };
};

手动执行

module.exports = (app) => {
  app.beforeStart(async () => {
    // 保证应用启动监听端口前，数据已经准备好
    // 后续数据的更新由定时任务自动触发
    await app.runSchedule('update_cache');
  });
};