@ fastify/kafka (法语)
快速化插件以互动 阿帕奇·卡夫卡支持卡夫卡生产商和消费者.
为了实现最佳性能,插件使用 @platformatic/kafka。 。 。 .
安装
npm i @fastify/kafka
兼容性
| 插件版本 | 快速化版本 |
|---|---|
>=3.x | ^5.x |
>=0.x <3.x | ^4.x |
>=0.x <3.x | ^3.x |
>=0.x <3.x | ^2.x |
>=0.x <3.x | ^1.x |
请注意,如果一个快取版本没有支持,那么该插件的相应版本也一样 上表列出. 见 快速化 LTS 政策 更多细节.
使用量
const crypto = require('node:crypto')
const fastify = require('fastify')()
const group = crypto.randomBytes(20).toString('hex')
fastify
.register(require('@fastify/kafka'), {
producer: {
bootstrapBrokers: ['127.0.0.1:9092'],
clientId: 'my-producer',
allowAutoTopicCreation: true
},
consumer: {
bootstrapBrokers: ['127.0.0.1:9092'],
groupId: group,
clientId: 'my-consumer',
allowAutoTopicCreation: true
}
})
fastify.post('/data', (req, reply) => {
fastify.kafka.push({
topic: 'updates',
payload: req.body,
key: 'dataKey'
})
reply.send({ ok: true })
})
fastify.kafka.subscribe('updates')
fastify.kafka.on('updates', (msg, commit) => {
console.log(msg.value.toString())
commit()
})
fastify.listen({ port: 3000 }, err => {
if (err) throw err
console.log(`Server listening on ${fastify.server.address().port}`)
fastify.kafka.consume()
})
更多如何使用此插件的例子, 您可以查看 示例目录。 。 。 .
从 node-rdkafka
从版本开始 3.x,此插件使用 @platformatic/kafka 改为 node-rdkafka。 。 。 .
配置选项的主要差异是:
node-rdkafka (旧) | @platformatic/kafka (新设) |
|---|---|
metadata.broker.list | bootstrapBrokers: ['host:port'] |
group.id | groupId |
fetch.wait.max.ms | maxWaitTime |
enable.auto.commit | autocommit |
auto.offset.reset | 通过 consumerTopicConf |
dr_cb | 无需 |
API 密码
这个模块暴露出以下API:
插件选项
{
producer?: Record<string, unknown> // @platformatic/kafka Producer options
consumer?: Record<string, unknown> // @platformatic/kafka Consumer options
producerTopicConf?: Record<string, unknown>
consumerTopicConf?: Record<string, unknown>
metadataOptions?: Record<string, unknown>
}
制作人
fastify.kafka.producer- 生产者实例fastify.kafka.push(message)- 向一个专题发送信息
fastify.kafka.push({
topic: 'my-topic', // required
payload: 'hello', // required - string or Buffer
key: 'myKey', // optional
partition: 0 // optional
})
消费者
fastify.kafka.consumer- 消费者案例fastify.kafka.subscribe(topic | topic[])——订阅一个或多个课题fastify.kafka.consume()——开始流出消息(流程模式)fastify.kafka.consume(n, callback)- 获取准确的n信件( 批量模式)fastify.kafka.on(topic, handler)- 听取关于特定主题的信息
// Flow mode
fastify.kafka.subscribe(['topic-a', 'topic-b'])
fastify.kafka.on('topic-a', (msg, commit) => {
console.log(msg.value.toString())
commit() // manually commit the offset
})
fastify.kafka.consume()
// Batch mode
fastify.kafka.subscribe('topic-a')
fastify.kafka.consume(10, (err, messages) => {
if (err) throw err
messages.forEach(msg => console.log(msg.value.toString()))
})
承认
这个项目由下列各方慷慨赞助:
过去提案国:
- 莱兹
许可证
许可证根据 麻省理工学院。 。 。 .