下面代碼只是消費信息的
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的消息會丟失,只需要處理重復消息就可以了。