# @sourceloop/task-service 中文文档

规范地址：https://npmcn.com/package/%40sourceloop/task-service/

版本：11\.1\.2
npm 元数据更新时间：2026-11-02T11:31:25.297Z
中文译文更新时间：2026-11-02T13:52:59.011Z
软件包许可：MIT

本文由上游 README 机器翻译，可能存在偏差，请结合原文核对 API、命令与配置。版权与许可归原作者所有。

原始 npm 页面：https://www.npmjs.com/package/%40sourceloop/task-service
原文对照：https://npmcn.com/package/%40sourceloop/task-service/?lang=original
项目主页：https://github.com/sourcefuse/loopback4-microservice-catalog#readme
源码仓库：https://github.com/sourcefuse/loopback4-microservice-catalog

## 关键信息（来自 npm 元数据）

- 包体积（解压后）：403 kB，364 个文件
- 模块格式：CommonJS
- TypeScript 类型：包内自带
- Node.js 要求：22 \|\| 24
- 运行依赖：13 个
- peer 依赖：@loopback/boot ^8\.0\.4，@loopback/core ^7\.0\.3，@loopback/rest ^15\.0\.4，@loopback/context ^8\.0\.3，@loopback/openapi-v3 ^11\.0\.4，@loopback/repository ^8\.0\.3，@loopback/rest-explorer ^8\.0\.4，@loopback/service-proxy ^8\.0\.3
- 首次发布：2023-10-05T16:13:22.971Z
- 最新版本发布：2026-11-02T11:31:24.925Z
- 已发布版本数：50（近 12 个月 10 个）

## 中文简介

可重用的、可定制的、基于工作流的任务服务

## 安装

```sh
npm install @sourceloop/task-service
```

也可以使用：`pnpm add @sourceloop/task-service`、`yarn add @sourceloop/task-service`、`bun add @sourceloop/task-service`

## README

[![ARC By SourceFuse logo](https://github.com/sourcefuse/loopback4-microservice-catalog/blob/master/docs/assets/logo-dark-bg.png?raw=true "ARC By SourceFuse")](https://sourcefuse.github.io/arc-docs/arc-api-docs)

# [@sourceloop /任务服务](https://github.com/sourcefuse/loopback4-microservice-catalog/tree/master/services/task-service)

 [![License](https://img.shields.io/github/license/sourcefuse/loopback4-microservice-catalog)](https://npmcn.com/package/%40sourceloop/task-service/LICENSE)[![Pb Loopback](https://img.shields.io/badge/Powered%20by-Loopback%204-brightgreen)](https://loopback.io/)

## 概述

一个可重用的、可定制的、基于工作流的任务服务，它根据分布式系统中通过不同微服务发生的各种事件创建一些可操作的任务。

要开始使用此服务的基本实现，请参见 [沙盒/ task-ms-example](https://github.com/sourcefuse/loopback4-microservice-catalog/tree/master/sandbox/task-example).

## 安装

```bash
npm i @sourceloop/task-service
```

## 开始使用

你可以开始使用 `@sourceloop/task-service` 只需4步：

1.  [配置并绑定TaskServiceComponent](https://npmcn.com/package/%40sourceloop/task-service/#readme-bind-component)
2.  [配置数据源](https://npmcn.com/package/%40sourceloop/task-service/#readme-configure-datasources)
3.  [运行迁移](https://npmcn.com/package/%40sourceloop/task-service/#readme-migrations)

## 绑定组件

-   选择并绑定一个传入连接器和一个传出连接器。参考 [输入和输出连接器部分了解更多详细信息](https://npmcn.com/package/%40sourceloop/task-service/#readme-incoming-and-outgoing-connectors)
-   绑定任务服务的配置（可选）-

```ts
this.bind(TaskServingBindings.Config).to({
  useCustomSequence: true, // enable this if you want to use your own sequence instead of one provided by the task service
  // though note that using a custom sequence may break or completely disable the authentication and authorization implemenation of task service
});
```

## 配置数据源

-   要使用任务服务，您需要绑定至少两个数据源（尽管它们都可以连接到同一个数据库）。
-   这个 `dataSourceName` 这两个的性质应该是 `WorkflowServiceSourceName` 和 `TaskDbSourceName` 任务服务导出的变量。例如，其中一个数据源看起来像这样-

```ts
import {inject, lifeCycleObserver, LifeCycleObserver} from '@loopback/core';
import {juggler} from '@loopback/repository';
import {TaskDbSourceName} from '@sourceloop/task-service';
import {config} from './config';

// Observe application's life cycle to disconnect the datasource when
// application is stopped. This allows the application to be shut down
// gracefully. The `stop()` method is inherited from `juggler.DataSource`.
// Learn more at https://loopback.io/doc/en/lb4/Life-cycle.html
@lifeCycleObserver('datasource')
export class PgDbDataSource
  extends juggler.DataSource
  implements LifeCycleObserver
{
  static dataSourceName = TaskDbSourceName; // this is the line that should variable from task service
  static readonly defaultConfig = config;

  constructor(
    @inject(`datasources.config.${TaskDbSourceName}`, {optional: true})
    dsConfig: object = config,
  ) {
    super(dsConfig);
  }
}
```

## 运行迁移

-   要运行任务服务提供的迁移(可在源代码的迁移文件夹和您的 `node_modules`)，使用 [db-migrate](https://db-migrate.readthedocs.io/en/latest/) 包中。
-   只有在计划使用HTTP传出连接器时才运行HTTP迁移。
-   此外，现在有一个选项可以选择SQL迁移或PostgreSQL迁移。 注意：对于@sourceloop/cli用户，这个选项可以在搭建过程中通过选择“数据源类型”选项来指定。

## 命令

由任务服务运行的工作流可以具有用于各种操作的服务任务，但是任务服务提供了两个常见的服务任务

#### 创建任务

这个命令需要一个包含任务列表的输入变量。该命令创建该变量中的所有任务，并使用创建的任务触发一个传出事件。要在节点上触发此命令，请使用主题- `create-tasks` 并提供一个结构为-的变量

```json
{
  "tasks": {
    "values": [
      {
        // task data
        ...
      }
    ]
  }
}

```

#### 结束任务

此命令预计将用作任务工作流结束事件的主题。例如，在camunda的情况下，它将是topic For [消息结束事件](https://docs.camunda.org/manual/7.20/reference/bpmn20/events/message-events/#message-end-event) 节点。要使用此命令，请使用topic - `end-task`

## 输入和输出连接器

任务服务需要 `IIncomingConnector` 和一个 `IOutgoingConnectors` 实现从外部源发送和接收事件。默认情况下，任务服务附带两组不同的连接器。注意，你可以使用不同类型的传入和传出连接器（例如，传入事件通过Kafka接收，但传出事件通过Http发送给webhook订阅）。

#### 卡夫卡

这组连接器实现了通过Kafka接收和发送事件的连接器。它可以被绑定为传入和传出连接器，并且需要一个额外的适配器绑定，以使kafka事件适应任务服务所期望的类型。你需要安装 [kafkajs](https://kafka.js.org/) 要使用这些连接器- `npm i kafkajs`

```ts
// Bind the config for Kafka connectors
this.bind(TaskServiceKafkaModule.CONFIG).to({
  // this part is required for both incoming and outgoing connectors
  connection: {
    clientId: process.env.KAFKA_CLIENT_ID,
    brokers: [...(process.env.KAFKA_SERVER?.split(',') ?? ['localhost:9092'])],
    ...(process.env.KAFKA_SSL === 'true'
      ? {
          ssl: true,
        }
      : {}),
  },
  // this part is required only if you use it as an incoming connector
  consumer: {
    groupId: process.env.KAFKA_GROUP_ID ?? 'task-service',
  },
  topics: ['sow.update'], // topics for receiving events
  // this part is required only if you use it as an outgoing connector
  producer: {},
  output: {
    topic: 'test', // topic for output events
  },
});

// bind the connector as an incoming connector if required
this.bind(TaskServiceBindings.INCOMING_CONNECTOR).toClass(KafkaStreamService);
// bind the connector as an outgoing connector if required
this.bind(TaskServiceBindings.OUTGOING_CONNECTOR).toClass(KafkaStreamService);
// bind the default adapter (it is available in `@sourceloop/task-service/kafka`)
this.bind(TaskServiceKafkaModule.ADAPTER).toClass(KafkaEventAdapter);
```

#### HTTP

这组连接器实现了通过HTTP调用发送和接收事件的代码。

```ts
// this line binds both the incoming and outgoing connectors plus some controllers required by the both
this.component(TaskHttpComponent);
// you can override either of the connectors by adding a new binding for them after the above
```

##### 传入的连接器

-   它通过端点-接收事件 `/events/trigger`
-   这个端点的有效载荷看起来像这样

```json
{
  "key": "event",
  "payload": {
    "name": "test",
    "description": "description"
  },
  "source": "test-source",
  "description": "description"
}
```

##### 即将离任的连接器

-   传出连接器向所有webhook订阅发布事件
-   webhook订阅是通过点击 `/webhooks/subscribe` 端点的有效载荷看起来像这样

```json
{
  "url": "http://localhost:3000", // the url that will hit with the payload for every outgoing event
  "key": "event-key" // the event keys for which this url would be hit
}
```

和一些必需的请求头 `x-api-key` 和 `x-api-secret`.

-   这些标头的值是通过另一个端点生成的 `/client-apps`。该端点应该被每个新客户端访问一次，并返回用于发送和验证webhook调用的新生成的密钥和秘密。呼唤 `/client-apps` 期望以下主体-

```json
{
  "clientName": "dummyName"
}
```

-   每个webhook调用也发送两个头 `x-task-signature` 和 `x-task-timestamp` 帮助验证客户端webhook调用的真实性。客户端可以通过使用事件有效负载和时间戳生成HMAC来验证此签名。下面给出了一个关于如何做到这一点的示例node.js代码

```js
function validateSignature(request) {
  const signature = request.headers['x-task-signature'];
  const timestamp = request.headers['x-task-timestampt'];
  const payload = request.body;
  // the secret in this line is the one generated by the /client-apps endpoint
  const hmac = createHmac('sha256', yourApiSecret);
  hmac.update(`${JSON.stringify(event)}:${timestamp}`);
  const expectedSignature = hmac.digest('hex');
  if (
    // compare both the strings
    !crypto.timingSafeEqual(
      Buffer.from(expectedSignature),
      Buffer.from(signature),
    )
  ) {
    // throw an error if signature does not match
    throw new HttpErrors.Unauthorized(INVALID_WEBHOOK_SIGNATURE);
  }
  return true;
}
```

### 与Sequelize一起使用

此服务支持Sequelize作为底层ORM使用 [@loopback / sequelize](https://www.npmjs.com/package/@loopback/sequelize) 扩展。为了使用它，您需要做以下更改。

1.要在应用程序中使用Sequelize，请在application.ts中添加以下内容：

```ts
this.bind(TaskServiceBindings.CONFIG).to({
  useCustomSequence: false,
  useSequelize: true,
});
```

2.  使用 `SequelizeDataSource` 在审计数据源中作为父类。参考 [这](https://www.npmjs.com/package/@loopback/sequelize#step-1-configure-datasource) 了解更多详情。

### 非对称令牌签名和验证

如果您正在使用非对称令牌签名和验证，则应该在服务中提供认证数据源，并在facade级别上提供认证数据源。auth数据库的数据源文件示例为：-

验证DB数据源：-

```ts
import {inject, lifeCycleObserver, LifeCycleObserver} from '@loopback/core';
import {juggler} from '@loopback/repository';
import {AuthDbSourceName} from '@sourceloop/core';
const DEFAULT_MAX_CONNECTIONS = 25;
const DEFAULT_DB_IDLE_TIMEOUT_MILLIS = 60000;
const DEFAULT_DB_CONNECTION_TIMEOUT_MILLIS = 2000;

const config = {
  name: 'auth',
  connector: 'postgresql',
  host: process.env.DB_HOST,
  port: process.env.DB_PORT,
  user: process.env.DB_USER,
  schema: process.env.DB_SCHEMA,
  password: process.env.DB_PASSWORD,
  database: process.env.AUTH_DB,
};

// Observe application's life cycle to disconnect the datasource when
// application is stopped. This allows the application to be shut down
// gracefully. The `stop()` method is inherited from `juggler.DataSource`.
// Learn more at https://loopback.io/doc/en/lb4/Life-cycle.html
@lifeCycleObserver('datasource')
export class AuthDataSource
  extends juggler.DataSource
  implements LifeCycleObserver
{
  static dataSourceName = AuthDbSourceName;

  static readonly defaultConfig = config;

  constructor(
    @inject('datasources.config.auth', {optional: true})
    dsConfig: object = config,
  ) {
    if (!!+(process.env.ENABLE_DB_CONNECTION_POOLING ?? 0)) {
      const dbPool = {
        max: +(process.env.DB_MAX_CONNECTIONS ?? DEFAULT_MAX_CONNECTIONS),
        idleTimeoutMillis: +(
          process.env.DB_IDLE_TIMEOUT_MILLIS ?? DEFAULT_DB_IDLE_TIMEOUT_MILLIS
        ),
        connectionTimeoutMillis: +(
          process.env.DB_CONNECTION_TIMEOUT_MILLIS ??
          DEFAULT_DB_CONNECTION_TIMEOUT_MILLIS
        ),
      };

      dsConfig = {...dsConfig, ...dbPool};
    }

    super(dsConfig);
  }
}
```

验证Cache Redis数据源：-

```ts
import {inject, lifeCycleObserver, LifeCycleObserver} from '@loopback/core';
import {AnyObject, juggler} from '@loopback/repository';
import {readFileSync} from 'fs';
import {AuthCacheSourceName} from '@sourceloop/core';

const config = {
  name: process.env.REDIS_NAME,
  connector: 'kv-redis',
  host: process.env.REDIS_HOST,
  port: process.env.REDIS_PORT,
  password: process.env.REDIS_PASSWORD,
  db: process.env.REDIS_DATABASE,
  url: process.env.REDIS_URL,
  tls:
    +process.env.REDIS_TLS_ENABLED! && process.env.REDIS_TLS_CERT
      ? {
          ca: readFileSync(process.env.REDIS_TLS_CERT),
        }
      : undefined,
  sentinels:
    +process.env.REDIS_HAS_SENTINELS! && process.env.REDIS_SENTINELS
      ? JSON.parse(process.env.REDIS_SENTINELS)
      : undefined,
  sentinelPassword:
    +process.env.REDIS_HAS_SENTINELS! && process.env.REDIS_SENTINEL_PASSWORD
      ? process.env.REDIS_SENTINEL_PASSWORD
      : undefined,
  role:
    +process.env.REDIS_HAS_SENTINELS! && process.env.REDIS_SENTINEL_ROLE
      ? process.env.REDIS_SENTINEL_ROLE
      : undefined,
};

@lifeCycleObserver('datasource')
export class RedisDataSource
  extends juggler.DataSource
  implements LifeCycleObserver
{
  static readonly dataSourceName = AuthCacheSourceName;
  static readonly defaultConfig = config;

  constructor(
    @inject(`datasources.config.${process.env.REDIS_NAME}`, {optional: true})
    dsConfig: AnyObject = config,
  ) {
    if (
      +process.env.REDIS_HAS_SENTINELS! &&
      !!process.env.REDIS_SENTINEL_HOST &&
      !!process.env.REDIS_SENTINEL_PORT
    ) {
      dsConfig.sentinels = [
        {
          host: process.env.REDIS_SENTINEL_HOST,
          port: +process.env.REDIS_SENTINEL_PORT,
        },
      ];
    }
    super(dsConfig);
  }
}
```

## 常见问题

### @sourceloop/task-service 是什么？

可重用的、可定制的、基于工作流的任务服务

### 如何安装 @sourceloop/task-service？

在项目目录运行 npm install @sourceloop/task-service，或使用 pnpm add / yarn add / bun add。它还需要宿主项目提供 peer 依赖：@loopback/boot（^8\.0\.4）、@loopback/core（^7\.0\.3）、@loopback/rest（^15\.0\.4）、@loopback/context（^8\.0\.3）、@loopback/openapi-v3（^11\.0\.4）、@loopback/repository（^8\.0\.3）、@loopback/rest-explorer（^8\.0\.4）、@loopback/service-proxy（^8\.0\.3）。npm 7 及以上与 pnpm 8 及以上默认会自动安装必需的 peer 依赖，Yarn 需要手动安装。

### @sourceloop/task-service 支持 TypeScript 吗？

支持。包内自带 TypeScript 类型声明（package\.json 声明了 types 入口），安装后即可获得类型提示，无需再安装 @types 包。

### @sourceloop/task-service 支持 ESM 还是 CommonJS？

以 CommonJS 形式发布，可以直接 require\(\)；在 ES Module 项目中也可以用 import 引入（Node\.js 会把 module\.exports 作为默认导出）。

### @sourceloop/task-service 需要什么版本的 Node\.js？

package\.json 的 engines 字段要求 Node\.js 22 \|\| 24。版本不符时 npm 默认只给出警告，开启 engine-strict 后会拒绝安装。

### @sourceloop/task-service 可以商用吗？

@sourceloop/task-service 使用 MIT 许可。宽松许可：允许商业使用、修改和再分发，通常只需在分发时保留原版权与许可声明。以上为概括，不构成法律意见，请以许可原文为准。

### @sourceloop/task-service 还在维护吗？

最新版本 11\.1\.2 发布于 2026/11/02（1 天前）；至今共发布 50 个版本，其中过去 12 个月发布了 10 个；首次发布于 2023/10/06。发布频率只是参考信号：功能稳定的工具可能很少需要更新，建议结合源码仓库的 issue 和提交记录判断。