kafka-node的學習使用


kafka-node地址:

下面代碼只是消費信息的

const kafka = require("kafka-node");
const Client = kafka.KafkaClient;
const Offset = kafka.Offset;
const Consumer = kafka.Consumer;

function toKafka() {
  const client = new Client({ kafkaHost: "192.168.100.129:9092" });
  const offset = new Offset(client);
  console.log("連接kafka中");

  const topics = [
    {
      topic: "test",
      partition: 0,
      offset: 0,
    },
  ];

  const options = {
    //  自動提交配置   (false 不會提交偏移量,每次都從頭開始讀取)
    autoCommit: true,
    autoCommitIntervalMs: 5000,
    //  如果設置為true,則consumer將從有效負載中的給定偏移量中獲取消息
    fromOffset: false,
  };

  const consumer = new Consumer(client, topics, options);

  consumer.on("message", function(message) {
    console.log(message);
  });

  consumer.on("error", function(message) {
    console.log(message);
    console.log("kafka錯誤");
  });
  //addTopics(topics, cb, fromOffset) fromOffset:true, 從指定的offset中獲取消息,否則,將從主題最后提交的offset中獲取消息。
  // consumer.addTopics(
  //   [{ topic: "", offset: 1 }],
  //   function(err, data) {
  //     console.log("根據指定的偏移量獲取值" + data);
  //   },
  //   true,
  // );

  consumer.on("offsetOutOfRange", function(topic) {
    topic.maxNum = 2;
    offset.fetch([topic], function(err, offsets) {
      const min = Math.min.apply(null, offsets[topic.topic][topic.partition]);
      consumer.setOffset(topic.topic, topic.partition, min);
    });
  });
  console.log("連接kafka后");
}

module.exports = toKafka;

估計后面再真正生產中可能會出現別的問題,后續在接着記錄

  最近幾天出現了重復消費信息和消息丟失的情況。主要原因是因為node.js是異步的,而且出現錯誤就導致項目停止,這樣被消費的消息就沒辦法提交。還有就是會出現重復的offset信息。本身對kafka就不是很懂,kafka-node的文檔說明也比較少,實在是想不到用什么好方法來解決這個問題,就是用了笨方法來解決了。下面是主要的代碼實現

//kafka 關鍵代碼

//定義上次消費的偏移量
let parkTopicsNum = parkTopicNum;
let userTopicsNum = userTopicNum;
const options = {
    // Auto commit config
    //  自動提交配置   (false 不會提交偏移量,每次都從頭開始讀取)
    autoCommit: false,
    autoCommitMsgCount: 100,
    autoCommitIntervalMs: 1000,
    // Fetch message config
    fetchMaxWaitMs: 100,
    fetchMinBytes: 1,
    fetchMaxBytes: 1024 * 10,
    //  如果設置為true,則consumer將從有效負載中的給定偏移量中獲取消息
    fromOffset: true,
    fromBeginning: false,
  };

  //手動設置從哪個偏移量開始拉取消息
  const consumer = new Consumer(client, topics, options);
  consumer.addTopics(
    [{ topic: KFK.parkTopics, offset: parkTopicNum + 1 }, { topic: KFK.userTopics, offset: userTopicNum + 1 }],
    function(err, added) {},
    true,
  );

  consumer.on("message", async function(message) {
    log.info(message);
    if (global.kafkaMsg === false) {
      global.kafkaMsg = true;
      if (message.topic === "parkDomainDataChanged") {
        if (message.offset > parkTopicsNum) {
          parkTopicsNum += 1;
          batis.query("apiGoalkeeper.insretKafka", {
            param: message.value,
            topic: message.topic,
            offset: message.offset,
            cTime: new Date(),
          });
          await updateMsg(message);
        }
      } else if (message.topic === "userDomainDataChanged") {
        if (message.offset > userTopicsNum) {
          userTopicsNum += 1;
          batis.query("apiGoalkeeper.insretKafka", {
            param: message.value,
            topic: message.topic,
            offset: message.offset,
            cTime: new Date(),
          });
        }
      }
      global.kafkaMsg = false;
    }
  });


//項目初始化關鍵代碼  根據topic查詢數據庫中最大的offset
  const parkTopics = (await batis.query("apiGoalkeeper.selectKafka", { topic: KFK.parkTopics }))[0].offset;
  const userTopics = (await batis.query("apiGoalkeeper.selectKafka", { topic: KFK.userTopics }))[0].offset;
  toKafka(parkTopics, userTopics);

//全局鎖定義
global.kafkaMsg = false;

主要的思路是:將每次kafka的消息存放在數據庫里面。在初始化的時候先查詢topic的最大offset,然后手動設置查詢到的offset+1開始拉取消息。每次拉取到的消息都會和上一次消費的偏移量對比,只有大於上一次偏移量的才會進行處理。這樣能保證不會丟失消息,也不會報錯導致項目停止,下次啟動的時候還是消費之前的消息。 這里使用了加鎖,可以保證無論來了多少kefka的消息,只要上一個消息沒有處理完成,下面的消息就沒辦法進入。

 consumer.on("message", async function(message) {
    log.info(message)
    if (message.topic === "parkDomainDataChanged") {
      if (message.offset > parkTopicsNum) {
        parkTopicsNum += 1;
        await batis.query("apiGoalkeeper.insretKafka", {
          param: message.value,
          topic: message.topic,
          offset: message.offset,
          cTime: new Date(),
        });
        await updateMsg(message);
      }
    } else if (message.topic === "userDomainDataChanged") {
      if (message.offset > userTopicsNum) {
        userTopicsNum += 1;
        await batis.query("apiGoalkeeper.insretKafka", {
          param: message.value,
          topic: message.topic,
          offset: message.offset,
          cTime: new Date(),
        });
      }
    }
  });

consumer.on這個方法只要有消息進入貌似就代表被消費了,我因為加鎖了,在鎖還還沒有開放的時候,有消息進入默認被消費了,等鎖開放的時候,此消息就沒辦法在獲取到了。
最終解決方案
  但是這樣還是出了問題,最終的解決方案是使用隊列做,也因為我們只是消費方,所以不需要管消息丟失,只需要處理重復消息就可以了。消息隊列是我們公司全棧寫的一個npm庫

const KFK = require("../../config").kafka;
const log = require("../../src/tools/Log");
const kafka = require("kafka-node");
const Client = kafka.KafkaClient;
const Offset = kafka.Offset;
const Consumer = kafka.Consumer;
const TaskQueue = require("@wolfx/async-task-queue");
const taskQueue = new TaskQueue();

function toKafka() {
  const client = new Client({ kafkaHost: KFK.kafkaUrl });
  const offset = new Offset(client);
  console.log("連接kafka成功");
  const topics = [
    {
      topic: KFK.userTopics,
      partition: 0,
      offset: 0,
    },
    {
      topic: KFK.parkTopics,
      partition: 0,
      offset: 0,
    },
  ];
  const options = {
    //  自動提交配置   (false 不會提交偏移量,每次都從頭開始讀取)
    autoCommit: true,
    autoCommitIntervalMs: 5000,
    //  如果設置為true,則consumer將從有效負載中的給定偏移量中獲取消息
    fromOffset: false,
  };
  
  const consumer = new Consumer(client, topics, options);

  consumer.on("message", async function(message) {
    //進來的消息進入隊列,這樣無論多少消息都會進入隊列等待,
    taskQueue.add(async () => {
      //從庫中查詢該消息是否已經被消費過
      const offset = (await batis.query("kafka.selectOffset", { topic: message.topic, offset: message.offset }))[0]
        .offset;
      if (offset < 1) {
        //只有為消費過的消息才會進行業務處理
        log.info("kafka ", message);
      }
    });
  });
  consumer.on("error", function(message) {
    console.log("kafka錯誤", message);
  });
}

module.exports = toKafka;

上面也說過consumer.on這個方法只要有消息進入就代表被消費了,雖然上面加鎖了,但是會出現消息來了,但是業務邏輯被鎖鎖住,后面不停的來消息,這樣就會丟失消息。這種存放在隊列中,不管有多少消息進來,都會進入到隊列里面,不需要擔心進入consumer.on的消息會丟失,只需要處理重復消息就可以了。


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM