# @comparaonline/event-streamer 中文文档

规范地址：https://npmcn.com/package/%40comparaonline/event-streamer/

版本：8\.2\.7
npm 元数据更新时间：2025-12-30T14:00:30.835Z
中文译文更新时间：2026-11-10T14:08:09.225Z
软件包许可：MIT

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

原始 npm 页面：https://www.npmjs.com/package/%40comparaonline/event-streamer
原文对照：https://npmcn.com/package/%40comparaonline/event-streamer/?lang=original
项目主页：https://github.com/comparaonline/event-streamer#readme
源码仓库：https://github.com/comparaonline/event-streamer

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

- 包体积（解压后）：676 kB，34 个文件
- 模块格式：CommonJS
- TypeScript 类型：包内自带
- Node.js 要求：\>=14\.17\.0
- 运行依赖：1 个
- 首次发布：2018-03-06T18:51:09.721Z
- 最新版本发布：2024-03-07T19:41:02.461Z
- 近 30 天下载量：2144（2026-10-11 至 2026-11-09，npm 官方下载统计）
- 已发布版本数：78（近 12 个月 0 个）

## 中文简介

简单的事件流程框架

## 安装

```sh
npm install @comparaonline/event-streamer
```

也可以使用：`pnpm add @comparaonline/event-streamer`、`yarn add @comparaonline/event-streamer`、`bun add @comparaonline/event-streamer`

## README

# 事件串流器

## 说明

Event Streamer是一个基于kafka事件通信连接微服务的库.

这是一包 [卡夫卡镇](https://github.com/tulios/kafkajs) 简化与主题的连接、错误和主题(事件名称、事件代码).

这个库无意同时消耗两个不同的集群,但可以同时以两个或两个以上的集群来制作.

## 安装

线条 :

```
$ yarn add @comparaonline/event-streamer
```

## 初始化

在使用前,应该先初始化,叫做方法 `setConfig`

### 基本生产者配置

这样就足以产生任何信息

```ts
import { setConfig } from '@comparaonline/events-streamer'

setConfig({
  host: 'kafka:9092'
})
```

### 基本消费者配置

如果你还需要消耗 事件的唯一区别是 消费者群体id是 **所需经费**

```ts
import { setConfig } from '@comparaonline/events-streamer'

setConfig({
  host: 'kafka:9092',
  consumer: {
    groupId: 'my-group-id'
  }
})
```

### 完整配置

| Prop | 类型 | 说明 | 使用 |
| --- | --- | --- | --- |
| 东道主 | 字符串 | Kafka经纪人,也可以使用结肠来分离(例如: `kafka-broker-1:9092,kafka-broker-2:9092`(中文(简体) ) | **所需经费** |
| 生产商 | 对象 | 对象 | _选项_ |
| 制片人 额外住房 | 字符串 \[\] | 额外发送消息的主机 。 如果您需要向两个或多个不同的集群发送信息,此选项将有所帮助 | _选项_ |
| 制作人. 连接TTL | 编号 | 上次发送信件后连接会保持开放时间( 以毫秒计) | _选项_  
默认: 5000 |
| 制作者. 重试选项 | 重试选项( 对象) | kafka- 节点重试选项: 重试、 因素、 分钟出局、 最大出局和随机化 | _选项_ |
| 制作人. 压缩Type | 压缩版( KafkaJS) | 设置压缩类型 。 只有无,而且没有其他执行方案 | _选项_  
默认: 压缩Types. NONE |
| 制片人. 专有 | 布尔 | 设定信件只发送一次 . **警告语** | _选项_  
默认: 假 |
| 制片人. 分离者 | 默认参与者/ 远程参与者 | 设定信件将如何被分区 | _选项_  
默认: 遗留部分 |
| 消费者 | 对象 | 对象 | _需要开始消费_ |
| 消费者. groupId | 字符串 | 卡夫卡团体 id | **所需经费** |
| 消费者.战略 | 战略(“专题”/“逐一”) | 如果您要创建主题队列或处理单个队列中的所有信件, 请选择 .  
_专题_:每个话题都会有一个独有的队列,并根据队列大小从kafka获取消息. 当队列满了时, 它会停止为这个特定主题取取回, 当一些处理器结束时恢复它 . **如果队列大小太小, 可能会出现拉格**  
_一个接一个_: 所有信件都会在一个队列中处理, 它需要等待前一个信件处理器完成后才能开始处理新信件 | _选项_  
默认: “ 专题 ” |
| 消费者.max 微信公众号 | 数字/ “ 无限” | 设置全局队列大小 | _选项_  
默认: 20 |
| 消费者.max 微信公众号 | 对象( 键: 字符串, 值: 最大MessagesPerTopic) | 设置特定主题队列大小 。 您也可以对特定主题使用“ 无限” | _选项_  
默认: 空对象 |
| 调试 | 虚假或调试 | 根据调试级别增加库记录 | _选项_ 默认: 假 |
| 卡夫卡-JSLogs (英语) | kafkajs.log 级别 | 设置 kafkajs 连接、 承诺和流线日志 | _选项_ 默认 : kafkajs.loglevel. nothING |
| 仅测试 | 布尔 | 避免卡夫卡服务器通信, 而不是发送/接收信件, 它将允许单位测试的额外方法 | _选项_  
默认: 假 |

## 使用量

### 活动主题

事件主题,事件代码或事件名称的用意是允许向同一个主题发送多条消息,但以不同方式处理.

_生产_ 信件将以 JSON 字符串形式发送,并附加一个名为“ 代码” 的属性 。 这个 `code` 将是事件的主题。 主题将永远传递给上加梅尔Case替换 `-_` 并且将每个字符之后的第一个字符转换成上层。 缩写:

-   `my-event-name`
-   `my event name`
-   `my_event_name`
-   `myEventName`

这一切将变成 `MyEventName`

如果不提供某一主题,则该主题将转换为上CamelCase,并成为主题.

_已消费_ 带有无效的 JSON 信件将被忽略 。 如果信件是有效的 JSON , 但不 `code` 属性将只由全球听众处理 .

### 生产

事件流器会按要求创建并关闭kafka客户端,以避免保持开通的连接,这种连接会被再用到TTL(每次执行后TTL会续用). 制作的活动 `data` 对象作为 JSON 字符串与相应的主题合并为单个主题。 还会加上 `appName` 财务报告和审定财务报表 `createdAt` 属性

#### 应用名称

`appName` 是一个具有发送人名的一级属性. `event-streamer` 有设置应用程序的选项Name

1.  发送 `appName` 信件上的属性
2.  执行 `event-streamer` 初始化为 `appName` 属性
3.  读取 `consumerGroupId` 从 `event-streamer` 初始化并用作 `appName`
4.  读取 `process.env.HOSTNAME`。这是在 K8S 部署时自动设置的
5.  设定 `unknown` 字符串

#### 消息日期

每个消息都应该有属性 `createdAt`,它是ISO UTC格式上的信息创建日期. `event-streamer` 自动添加此属性

#### 单一活动

```ts
import { emit } from '@comparaonline/event-streamer'

// Remember to call setConfig before use this method

async function main () {
  await emit({
    topic: 'my-topic',
    eventName: 'my-event-name',
    data: { firstName: 'John' }
  })
  
  // Topic: my-topic Message: { "firstName": "John", "code": "MyEventName", "createdAt": "2024-03-07 18:41:54Z", "appName": "test" }

}
```

#### 带有替代语法的单个事件

```ts
import { emit } from '@comparaonline/event-streamer'

async function main () {
  await emit('my-topic', 'my-event-name', { firstName: 'John' })
  // Topic: my-topic Message: { "firstName": "John", "code": "MyEventName", "createdAt": "2024-03-07 18:41:54Z", "appName": "test" }

  await emit('my-topic', { firstName: 'John' })
  // Topic: my-topic Message: { "firstName": "John", "code": "MyTopic", "createdAt": "2024-03-07 18:41:54Z", "appName": "test" }
}
```

#### 就同一项主题的两个事件

```ts
import { emit } from '@comparaonline/event-streamer'

async function main () {
  await emit({
    topic: 'my-topic',
    eventName: 'my-event-name',
    data: [{ firstName: 'John' }, { firstName: 'Jane' }]
  })
  
  // Topic: my-topic Message: { "firstName": "John", "code": "MyEventName", "createdAt": "2024-03-07 18:41:54Z", "appName": "test" }
  // Topic: my-topic Message: { "firstName": "Jane", "code": "MyEventName", "createdAt": "2024-03-07 18:41:54Z", "appName": "test" }

}
```

#### 两个活动到两个不同的话题

```ts
import { emit } from '@comparaonline/event-streamer'

async function main () {
  await emit(
    [
      {
        topic: 'my-topic-a',
        eventName: 'my-event-name',
        data: { firstName: 'John' }
      },
      {
        topic: 'my-topic-b',
        eventName: 'my-event-name',
        data: { firstName: 'Jane' }
      }
    ]
  )
  
  // Topic: my-topic-a Message: { "firstName": "John", "code": "MyEventName", "createdAt": "2024-03-07 18:41:54Z", "appName": "test" }
  // Topic: my-topic-b Message: { "firstName": "Jane", "code": "MyEventName", "createdAt": "2024-03-07 18:41:54Z", "appName": "test" }
}
```

#### 无主题的单个事件

```ts
import { emit } from '@comparaonline/event-streamer'

async function main () {
  await emit({
    topic: 'my-topic',
    data: { firstName: 'John' }
  })
  
  // Topic: my-topic Message: { "firstName": "John", "code": "MyTopic", "createdAt": "2024-03-07 18:41:54Z", "appName": "test" }
}
```

#### 单个活动到不同的主机/集群

```ts
import { emit, setConfig } from '@comparaonline/event-streamer'

setConfig({
  host: 'kafka-gpc:9092',
  producer: {
    additionalHosts: ['kafka-aws:9092']
  }
})

async function main () {
  await emit({
    topic: 'my-topic',
    data: { firstName: 'John' }
  }, 'kafka-azure:9092')
  
  // Cluster: kafka-azure:9092
  // Topic: my-topic Message: { "firstName": "John", "code": "MyTopic" }
}
```

#### 主体不允许

```ts
import { emit } from '@comparaonline/event-streamer'

async function main () {
  await emit({
    topic: 'my-topic',
    data: { firstName: 'John', code: 'Code' }
  }) // Throw Error: 'Reserved object keyword "code" inside data'
}
```

```ts
import { emit } from '@comparaonline/event-streamer'

async function main () {
  await emit({
    topic: 'my-topic',
    data: { firstName: 'John' },
    eventName: ''
  }) // Throw Error: 'Invalid message code'
}
```

### 消费

你至少需要听一个话题。 重要的是要知道将要获取的消息的大小,因为这将决定同时处理多少消息. 缩写:

| Avg 消息 | 获取SizeInMB | 并行行动 |
| --- | --- | --- |
| 200千贝 | 0.5 (韩语) | 2个 |
| 300千贝 | 0.5 (韩语) | 1个 |
| 200千贝 | 3(默认) | 14/15 (中文(简体) ) |
| 600千贝 | 0.5 (韩语) | 消费者卡住 |

交付给处理器的数据对象( 第1个参数)将是一个包括 `code` 属性

交给处理器的 Emit 函数( 第二个参数) 是相同的 `emit` 函数用于生成事件,但不需要导入

```ts
import { ConsumerRouter } from '@comparaonline/event-streamer'

// Remember to call setConfig with a groupId

async function main () {
  const consumer = new ConsumerRouter()

  // This handler will be executed with every message from 'topic-a', even if they don't have 'code' property
  consumer.add('topic-a', (data, emit) => { console.log(1) })

  // This handler will be executed only with messages from 'topic-b' with property 'code' equal to 'EventNameB'
  consumer.add('topic-b', 'event-name-b', (data, emit) => { console.log(2) })

  // This handler will be executed with message from 'topic-c' with property 'code' equal to 'EventNameC1' or 'EventNameC2'
  consumer.add('topic-c', ['event-name-c-1', 'event-name-c-2'], (data, emit) => { console.log(3) }) 

  // This handler will be executed with every message from 'topic-d' or 'topic-e'
  consumer.add(['topic-d', 'topic-e'], (data, emit) => { console.log(4) })

  // This handler will be executed with messages from 'topic-e' or 'topic-f' with property 'code' equal to 'MyEventName'
  consumer.add(['topic-e', 'topic-f'], 'my-event-name', (data, emit) => { console.log(5) })

  // This handler will be executed with messages from 'topic-g' and 'topic-h' with property 'code' equal to 'MyEventName1' or 'MyEventName2'
  consumer.add(['topic-g', 'topic-h'], ['my-event-name-1', 'my-event-name-2'], (data, emit) => { console.log(6) })

  // Alternative syntax 

  consumer.add({
    topic: 'topic-i',
    eventName: 'my-event-name-i', // this is optional
    callback: (data, emit) => { console.log(6) }
  })
  
  await consumer.start()
}
```

#### 输入/ 输出示例

| 专题 | 代码 | 日志 |
| --- | --- | --- |
| 专题a | _未定义_ | 1个 |
| 专题a | "Topica" (意大利语) | 1个 |
| 专题a | "我的晚会名" | 1个 |
| 专题b | “活动名称A” | \-- -- . . |
| 专题b | 活动名称B | 2个 |
| 专题b | 托比克语 | \-- -- . . |
| 专题c | 事件Name C1 | 3个 |
| 专题c | 事件NameC2 | 3个 |
| 专题c | 事件NameC3 | \-- -- . . |
| 专题d | _未定义_ | 页:1 |
| 专题d | "TopicD" (中文(简体) ) | 页:1 |
| 专题d | "我的晚会名" | 页:1 |
| 专题e | _未定义_ | 页:1 |
| 专题e | "托皮克"(Topice) | 页:1 |
| 专题e | "我的晚会名" | 第4、第5条 |
| 专题f | _未定义_ | \-- -- . . |
| 专题f | "TopicF" (意大利语) | \-- -- . . |
| 专题f | "我的晚会名" | 页:1 |
| 专题-g | _未定义_ | \-- -- . . |
| 专题-g | "TopicG" (美国英语) | \-- -- . . |
| 专题-g | "My EventName 1" (英语) | 6个 |
| 专题-g | 我的晚会Name2 | 6个 |
| 专题h | _未定义_ | \-- -- . . |
| 专题h | "托皮克"(TopicH) | \-- -- . . |
| 专题h | "My EventName 1" (英语) | 6个 |
| 专题h | 我的晚会Name2 | 6个 |

## 测试

当您需要进行单位/ 整合测试时, 但您不打算连接一个真实的 kafka 服务器时, 您可以直接下线使用 .

要打这个电话 `setConfig` 与 `onlyTesting` 打开 **真实** 在你的笑话设置。 这将允许您使用额外的方法

### 设置

```ts
// src/test/setup/event-streamer.ts
import { setConfig } from '@comparaonline/event-streamer'

setConfig({
  host: 'any-host',
  consumer: {
    groupId: 'fake-group-id'
  },
  onlyTesting: true
})
```

### 事件服务器

```ts
// src/event-server/index.ts
import { ConsumerRouter } from '@comparaonline/event-streamer'
import { application } from '../application'

// Remember to call setConfig with a groupId

export const consumer = new ConsumerRouter()
consumer.add('topic-a', 'event-name-a', async (data, emit) => { 
  await emit({
    topic: 'topic-b',
    data: {
      message: 'Event received'
    }
  })
})

application.onStart(async () => {
  await consumer.start()
})
```

### 服务

```ts
// src/services/my-service/__tests__/index.test.ts
import { consumer } from '../../../event-server'
import { getEmittedEvents, clearEmittedEvents, getParsedEmittedEvents } from '@comparaonline/event-server'

describe('Testing some handlers', () => {
  beforeEach(() => {
    clearEmittedEvents()
  });

  describe('Testing EventNameA', () => {
    it('Should emit event into topic b', async () => {
      await consumer.input({
        data: { someData: true },
        topic: 'a',
        eventName: 'event-name-a'
      })

      const events = getEmittedEvents()

      expect(events[0]).toMatchObject({
        topic: 'topic-b',
        messages: [
          JSON.stringify({
            message: 'Event received',
            code: 'TopicB'
          })
        ]
      })
    })

    it('Should emit event into topic c with parsed event', async () => {
      await consumer.input({
        data: { someData: true },
        topic: 'a',
        eventName: 'event-name-a'
      })

      const events = getEmittedEvents()

      expect(events[0]).toMatchObject({
        topic: 'topic-b',
        eventName: 'TopicB',
        data: {
          message: 'Event received',
          code: 'TopicB'
        }
      })
    })
  })
})

```

## 迁移到版本+8.0.0

### 节点版本

您需要大于或等于 v14.17. 0 的节点版本 。 Docker 文件的好选择是 `node:14.17.0-alpine3.13` 至少比你已经想跟 `node:16.13-alpine3.15`

#### 更新节点后已知的问题

-   `pg`:如果您正在使用一个来自节点 v13 的上一个版本,并带有 pg v7. X. X. 那么它就不会再运行了. 最容易解决的方法是 `yarn add pg@8.0.3`
-   `python`: 在您的居克文件更改中 `python` 改为 `python3`

### 专题和议题

#### 听众

以前的版本读取了所有专题,并且执行与专题无关的每一事件的行动。 现在你需要链接主题/事件/处理器

```ts
// Option A: you know where the event came from
consumer.add('topic-a', 'my-event-name', myHandler)

// Option B: you don't know where the event came from
consumer.add(['topic-a', 'topic-b'], 'my-event-name', myHandler)

// Option C: you know where the event came from but there are two input events with the same action
consumer.add('topic-a', ['my-event-name-a', 'my-event-name-b'], myHandler)

// Option D: you don't know where the event came from but there are two input events with the same action
consumer.add(['topic-a', 'topic-b'], ['my-event-name-a', 'my-event-name-b'], myHandler)
```

#### 活动主题

上一个事件流媒体版本使用类名作为事件主题. 现在您需要指定它们, 无论事件主题会自动转换为上骆驼案件

#### 提示

-   您可以将事件文件夹转换为简单的 TS 接口
-   创建一个包含所有发送事件的 kafka 服务文件夹

### 移民公关

[Flux 比较器同步](https://github.com/comparaonline/flux-comparador-sync/pull/53/files)

[结果连接器](https://github.com/comparaonline/results-connector/pull/67/files)

[引用应用 CICL](https://github.com/comparaonline/quoteapp-cicl/pull/127/files)

## 常见问题

### @comparaonline/event-streamer 是什么？

简单的事件流程框架

### 如何安装 @comparaonline/event-streamer？

在项目目录运行 npm install @comparaonline/event-streamer，或使用 pnpm add / yarn add / bun add。

### @comparaonline/event-streamer 支持 TypeScript 吗？

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

### @comparaonline/event-streamer 支持 ESM 还是 CommonJS？

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

### @comparaonline/event-streamer 需要什么版本的 Node\.js？

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

### @comparaonline/event-streamer 可以商用吗？

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

### @comparaonline/event-streamer 的下载量有多少？

根据 npm 官方下载统计，@comparaonline/event-streamer 在 2026/10/11 – 2026/11/09 这 30 天内被下载 2,144 次，日均约 71 次。下载量会把 CI 构建、镜像同步和作为其他包依赖的安装一并计入，适合横向比较热度，不能直接等同于使用人数。

### @comparaonline/event-streamer 还在维护吗？

最新版本 8\.2\.7 发布于 2024/03/08（2 年前）；至今共发布 78 个版本，其中过去 12 个月发布了 0 个；首次发布于 2018/03/07。发布频率只是参考信号：功能稳定的工具可能很少需要更新，建议结合源码仓库的 issue 和提交记录判断。