nodejs中的kafkajs,消费顺序,不重复消费

Posted 江山一族

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了nodejs中的kafkajs,消费顺序,不重复消费相关的知识,希望对你有一定的参考价值。

参考:https://kafka.js.org/docs

1.封装kafkaUtil类



const { Kafka, logLevel } = require(‘kafkajs‘) //const cache = require(‘../conn/redis.js‘); const kafka = new Kafka({ clientId: ‘my-app‘, brokers: [ "lcoalhost:8092", "localhost:8093", "localhost:8094", "lcoalhost:8095", "localhost:8096", ], retry: { retries: 8 }, logLevel: logLevel.ERROR }) /** * 如果groupId已存在重复的,建立不同的kafka实例会报错 */ /** * kafka生产者发送消息 * messages: [{ value: ‘Hello KafkaJS user!‘, }, { value: ‘Hello KafkaJS user2!‘, }], */ exports.producer = async (topic, groupId, msg) => { try { const producer = kafka.producer({ groupId: groupId }) await producer.connect() await producer.send({ topic: topic, messages: msg, acks: 1 }) } catch (error) { throw error; } } exports.consumer = async (topic, groupId, callback) => { try { const consumer = kafka.consumer({ groupId: groupId }) await consumer.connect() await consumer.subscribe({ topic: topic }) await consumer.run({ autoCommit: true, eachMessage: async ({ topic, partition, message }) => {
//防止重复消费数据 await consumer.commitOffsets([{ topic: topic, partition: partition, offset: Number(message.offset) + 1 }]) let msg = message.value.toString() console.log(72, ‘消费者接收到的数据为:‘, msg); callback(msg); } }) } catch (err) { throw err; } }

2.producer.js

   

const kafka = require(‘./kafkaUtil‘);
(async function () {
    const topic = ‘MY——TOPIC1‘
    const groupId = ‘MY——TOPIC1‘
    try {
        for (let i = 0; i < 10000; i++) {
            await new Promise((resolve, reject) => {
                setTimeout(async () => {
                    resolve(1)
                }, 1000)
            }).then(async () => {
                console.log(‘发送的数据为:‘, i)
                await kafka.producer(topic, groupId, [{
                    key: "a",//key值为了保证消费者按照生产者生产的数据顺序,消费数据,key值必须一致;如果不需要消费者按照生产的顺序消费,key去掉即可,参考: https://www.zhihu.com/question/266390197
                    value: `${i}`
                }])
            })
        }
    } catch (error) {
        console.log(14, error)
        throw error;
    }

})()

3.consumer.js

 

const kafka = require(‘./kafkaUtil‘);
(async function () {
    const fs = require(‘fs‘);
    let count = 1;
    const topic = ‘MY——TOPIC1‘
    const groupId = ‘MY——TOPIC1‘
    try {
        await kafka.consumer(topic, groupId, async (msg) => {
            let str = `第${count}接收到的数据为:${msg}`;
            count++;
            fs.writeFileSync(`${process.cwd()}/test01.txt`, str, {
                flag: ‘a‘,
            })
            console.log(str)
        })
    } catch (error) {
        console.log(14, error)
        throw error;
    }
})()

经实际测试,没有发现消费问题。如有发现问题,请多多指教,谢谢。。。  

 

以上是关于nodejs中的kafkajs,消费顺序,不重复消费的主要内容,如果未能解决你的问题,请参考以下文章

系统学习消息队列分享(七) 如何处理消费过程中的重复消息?

读取 kafka 主题并通过 Rest API 公开数据以供 prometheus 抓取(Nodejs)

如何解决 顺序消费重复消费

消息队列面试题要点

消息队列面试题要点

详解RocketMQ 顺序消费机制