egg-bus
TypeScript icon, indicating that this package has built-in type declarations

2.0.3 • Public • Published

egg-bus

NPM version build status Test coverage David deps Known Vulnerabilities npm download

🐣 用 egg 编写优雅的队列与事件,基于 bull 实现

安装

$ npm i egg-bus bull --save

使用

exports.bus = {
  enable: true,
  package: "egg-bus",
};

配置

// {app_root}/config/config.default.js
exports.bus = {
  agent: true, // 默认为false  若设置为true 则同时挂载在agent 上
  debug: true, // Debug 模式下会打印更多日志信息
  concurrency: 1, // Bull 中队列处理的并发数:https://github.com/OptimalBits/bull/blob/develop/REFERENCE.md#queueprocess
  listener: {
    ignore: null, // 忽略目录中的某些文件,https://eggjs.org/zh-cn/advanced/loader.html#ignore-string
    baseDir: "listener",
    options: {
      // Bull Job 配置: https://github.com/OptimalBits/bull/blob/develop/REFERENCE.md#queueadd
      attempts: 5,
      backoff: {
        delay: 3000,
        type: "fixed",
      },
    },
  },
  job: {
    // 与 listener 一致,唯一不同的就是 默认 baseDir 的值为 `job`
  },
  bull: {
    // Bull 队列配置:https://github.com/OptimalBits/bull/blob/develop/REFERENCE.md#queue
    redis: {
      host: "localhost",
      port: 6379,
      db: 0,
    },
  },

  queue: {
    default: "default", // 默认队列名称
    prefix: "bus", // 队列前缀
  },
  queues: {
    // 针对不同队列单独配置

    // 比如针对默认队列更改 redis 端口
    default: {
      concurrency: 2, // 针对队列单独设置并发
      redis: {
        port: 6380,
      },
    },
  },
};

更多配置说明请查看 config/config.default.js

使用

bus 会读取 app 目录中的 listenerjob 目录(默认,可通过配置修改)并解析 相关配置。

app
├── controller
│   ├── home.ts
├── job        <-- 队列目录
│   └── something.ts
├── listener   <-- 事件监听目录
│   ├── spy.ts
├── router.ts

Job

job 是队列中的一项任务。在 app/job 目录,定义一个 job

const { Job } = require("egg-bus");

class DemoJob extends Job {
  static get queue() {
    return "queue_name"; // 使用的队列名称
  }

  static get attempts() {
    return 5; // 重试次数
  }

  async run(data, job) {
    // job 任务运行时调用
    // 第一个参数是发送过来的数据
    // 第二个参数是 Bull 的原始 Job 对象
    // 通过 this.ctx 和 this.app 分别获取 egg 的 Context 和 Application 对象
  }

  failed(data) {
    // 当 job 失败并重试达到限定次数后调用
  }
}

module.exports = DemoJob;

在 run 中抛出任何未捕获的异常都会认为 job 执行失败,会在指定次数内重新尝试。

通过 dispatch 方法触发一个 job

const data = { name: "abel" };
app.bus.dispatch("demo", data);

Listener

listener 用于监听事件发生并执行某些任务。很多情况下,我们需要将核心业务与一些耗时 任务分开来提高响应速度或解耦增强可维护性和移植性。

nodejs 本身支持异步,可以解决响应速度的问题,基于事件,也可以达到解耦的目的。但实 际使用起来却有很多麻烦:

  1. 在异步中运行的代码如果不使用 app.runInBackground 就无法被 egg 捕获异常而记录日志;
  2. 如果发生错误,没有重试机制;
  3. nodejs 本身的 event 机制在 listener 数量上有限制,虽然可以通过参数提高这个阈值, 但这可能引发其它问题。
  4. 不够优雅!不够优雅!不够优雅!有很多类似的模块来解决这些问题,但大多只提供了基础功能。 比如告诉你怎么创建队列,怎么监听队列,却并没有告诉你这些创建队列、监听队列的代码应该放在何处。

因此,为了解决上面这些问题,参考 laravel 的事件机制设计了 listener

const { Listener } = require("egg-bus");

class DemoListener extends Listener {
  static get watch() {
    return ["opened", "visited"]; // 监听的事件名称
  }

  static get queue() {
    return "queue_name"; // 使用的队列名称
  }

  static get attempts() {
    return 5; // 重试次数
  }

  async run(event, job) {
    // listener 任务运行时调用
    // 与 job 不同,第一个参数是 event 对象,其中包含以下值:
    // - name 事件名称
    // - data 数据
    // 第二个参数是 Bull 的原始 Job 对象
    // 通过 this.ctx 和 this.app 分别获取 egg 的 Context 和 Application 对象
    console.log(event.name, event.data);
  }

  failed(event, error, job) {
    // 当 listener 失败并重试达到限定次数后调用
  }
}

module.exports = DemoListener;

事件的监听并不需要编写对应关系,你只需要告诉 listener 需要注意哪些事件就行了。

通过 emit 方法触发一个 事件

const data = { name: "abel" };
app.bus.emit("opened", data);

Api 参考

所有可用的 api 都位于 app.bus 对象下:

dispatch

触发一个任务

dispatch(name: string, payload?: any, options?: JobOptions): void
  • name 任务名,和 job 文件名一致
  • payload 发送的数据
  • options Bull Job 的一些定制化选项

emit

触发一个事件

emit(name: string, payload?: any, options?: JobOptions): void
  • name 事件名称
  • payload 发送的数据
  • options Bull Job 的一些定制化选项

get

获取原始的 Bull 队列实例

get(name: string): Queue
  • name 队列名称

问题和建议

请创建 issue.

License

MIT

Package Sidebar

Install

npm i egg-bus

Weekly Downloads

43

Version

2.0.3

License

MIT

Unpacked Size

19.1 kB

Total Files

15

Last publish

Collaborators

  • seekcx