kafka node使用記錄

前言：

專案設計使用kafka實現系統訊息的收發，前端則使用kafka-node來實現，由於之前開發比較倉促，對kafka-node高可用性一直沒有測試，故以此篇文章記錄一下開發及測試過程。

測試環境：軟體

版本kafka

kafka_2.10-0.10.1.1

kafka-node

2.2.3

nodejs

6.11.0

部署方式：

kafka使用三颱虛擬機器進行集群部署，其中在配置kafka服務相關引數時，在server.properties檔案中修改num.partitions=3default.replication.factor=3，以此建立3個partition分割槽以及每個分割槽建立3個副本。

**實現：

生產者

function
kafkasendmessage()
);const producer =
newproducer
(client,);
producer.on(
'ready'
,function()
;        producer.
send([
payload]
,function
(err, data)
else})
;});
producer.on(
'error'
,function
(err));
}

備註：第7行**中partitionertype表示使用partition方式，如不配置這項，生產者傳送訊息預設傳送到partition0中，無法達到多個partition負載均衡的效果。

該引數在kafka-node官方有介紹：

引數列舉值

說明使用場景

defaultpartitioner

0預設傳送到partition0

kafka服務中只有乙個partiton

randompartitioner

1隨機傳送到某個partition

每次生產者傳送訊息體中topic數量等於1

cyclicpartitioner

2迴圈傳送到partition(0 1 2)

每次生產者傳送訊息體中topic數量大於1

消費者

function
kafkarecievemessage()
);let topics =
;    topics.
push
('testtopic');
let offset =
newoffset
(client)
;    offset.
fetchlatestoffsets
(topics,
(event, offsetvalue)
=>
console.
log(topic1)
;                topics_partitions.
push
(topic1);}
}let options =
;let consumer =
newconsumer
(            client, topics_partitions, options
);        consumer.on(
'message'
,function
(message));
});}

備註：第10行**中使用offset獲取topic最新的偏移量及全部partition資料，第19行**輸出topictest的偏移量及partition資訊。

測試：

function
test()
,5000);
}

結果

offset
offset
offset
producer
}consumer
consumer
producer
}producer
}consumer
consumer
producer
}consumer
producer
}consumer
producer
}consumer
producer
}consumer
producer
}consumer
producer
}consumer
producer
}

備註：offset為獲取topictest所有partition及offset資料，producer每5秒隨機向某個broker中的所在的partition傳送資料，consumner輸出接受資料。經測試當停到kafka集群中某個節點時，客戶端依然可以正常收發，保證了kafka服務的高可用性。

kafka node使用記錄

Python 使用記錄（累計記錄）

fiddler使用記錄

xshell securecrt使用記錄

kafka node使用記錄

Python 使用記錄（累計記錄）

fiddler使用記錄

xshell securecrt使用記錄

相關推薦