事件串流器

说明

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

这是一包 卡夫卡镇 简化与主题的连接、错误和主题(事件名称、事件代码).

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

安装

线条 :

$ yarn add @comparaonline/event-streamer

初始化

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

基本生产者配置

这样就足以产生任何信息

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

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

基本消费者配置

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

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

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

完整配置

使用量

活动主题

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

生产 信件将以 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 自动添加此属性

单一活动

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" }

}

带有替代语法的单个事件

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" }
}

就同一项主题的两个事件

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" }

}

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

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" }
}

无主题的单个事件

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" }
}

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

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" }
}

主体不允许

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'
}

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

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

消费

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

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

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

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()
}

输入/ 输出示例

测试

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

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

设置

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

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

事件服务器

// 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()
})

服务

// 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 [email protected]
• python: 在您的居克文件更改中 python 改为 python3

专题和议题

听众

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

// 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 比较器同步

结果连接器

引用应用 CICL