摘要:消息以為類別記錄將消息種子分類每一類的消息稱之為一個主題。這意味著生產者不等待來自同步完成的確認繼續發送下一條批消息。這意味著在已成功收到的數據并得到確認后發送下一條。三種機制,性能依次遞減吞吐量降低,數據健壯性則依次遞增。
kafka 簡介
Kafka 是一種高吞吐量的分布式發布訂閱消息系統kafka角色必知
producer:生產者。 consumer:消費者。 topic: 消息以topic為類別記錄,Kafka將消息種子(Feed)分類, 每一類的消息稱之為一個主題(Topic)。 broker:以集群的方式運行,可以由一個或多個服務組成,每個服務叫做一個broker;消費者可以訂閱一個或多個主題(topic), 并從Broker拉數據,從而消費這些已發布的消息。經典模型
1. 一個主題下的分區不能小于消費者數量,即一個主題下消費者數量不能大于分區屬,大了就浪費了空閑了 2. 一個主題下的一個分區可以同時被不同消費組其中某一個消費者消費 3. 一個主題下的一個分區只能被同一個消費組的一個消費者消費常用參數說明 request.required.acks
Kafka producer的ack有3中機制,初始化producer時的producerconfig可以通過配置request.required.acks不同的值來實現。 0:這意味著生產者producer不等待來自broker同步完成的確認繼續發送下一條(批)消息。此選項提供最低的延遲但最弱的耐久性保證(當服務器發生故障時某些數據會丟失,如leader已死,但producer并不知情,發出去的信息broker就收不到)。 1:這意味著producer在leader已成功收到的數據并得到確認后發送下一條message。此選項提供了更好的耐久性為客戶等待服務器確認請求成功(被寫入死亡leader但尚未復制將失去了唯一的消息)。 -1:這意味著producer在follower副本確認接收到數據后才算一次發送完成。 此選項提供最好的耐久性,我們保證沒有信息將丟失,只要至少一個同步副本保持存活。 三種機制,性能依次遞減 (producer吞吐量降低),數據健壯性則依次遞增。auto.offset.reset
1. earliest:自動將偏移重置為最早的偏移量 2. latest:自動將偏移量重置為最新的偏移量(默認) 3. none:如果consumer group沒有發現先前的偏移量,則向consumer拋出異常。 4. 其他的參數:向consumer拋出異常(無效參數)kafka安裝和簡單測試 安裝kafka(不需要安裝,解包即可)
# 官方下載地址:http://kafka.apache.org/downloads # wget https://www.apache.org/dyn/closer.cgi?path=/kafka/1.1.1/kafka_2.12-1.1.1.tgz tar -xzf kafka_2.12-1.1.1.tgz cd kafka_2.12-1.1.0啟動kafka server
# 需先啟動zookeeper # -daemon 可啟動后臺守護模式 bin/zookeeper-server-start.sh config/zookeeper.properties bin/kafka-server-start.sh config/server.properties啟動kafka客戶端測試
# 創建一個話題,test話題2個分區 bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 2 --topic test Created topic "test". # 顯示所有話題 bin/kafka-topics.sh --list --zookeeper localhost:2181 test # 顯示話題信息 bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test Topic:test PartitionCount:2 ReplicationFactor:1 Configs: Topic: test Partition: 0 Leader: 0 Replicas: 0 Isr: 0 Topic: test Partition: 1 Leader: 0 Replicas: 0 Isr: 0 # 啟動一個生產者(輸入消息) bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test [等待輸入自己的內容 出現>輸入即可] >i am a new msg ! >i am a good msg ? # 啟動一個消費者(等待消息) # 注意這里的--from-beginning,每次都會從頭開始讀取,你可以嘗試去掉和不去掉看下效果 bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning [等待消息] i am a new msg ! i am a good msg ?安裝kafka的php擴展
# 先安裝rdkfka庫文件 git clone https://github.com/edenhill/librdkafka.git cd librdkafka/ ./configure make sudo make install git clone https://github.com/arnaud-lb/php-rdkafka.git cd php-rdkafka phpize ./configure make all -j 5 sudo make install vim [php]/php.ini extension=rdkafka.sophp代碼實踐 生產者
setDrMsgCb(function ($kafka, $message) { file_put_contents("./dr_cb.log", var_export($message, true).PHP_EOL, FILE_APPEND); }); $conf->setErrorCb(function ($kafka, $err, $reason) { file_put_contents("./err_cb.log", sprintf("Kafka error: %s (reason: %s)", rd_kafka_err2str($err), $reason).PHP_EOL, FILE_APPEND); }); $rk = new RdKafkaProducer($conf); $rk->setLogLevel(LOG_DEBUG); $rk->addBrokers("127.0.0.1"); $cf = new RdKafkaTopicConf(); // -1必須等所有brokers同步完成的確認 1當前服務器確認 0不確認,這里如果是0回調里的offset無返回,如果是1和-1會返回offset // 我們可以利用該機制做消息生產的確認,不過還不是100%,因為有可能會中途kafka服務器掛掉 $cf->set("request.required.acks", 0); $topic = $rk->newTopic("test", $cf); $option = "qkl"; for ($i = 0; $i < 20; $i++) { //RD_KAFKA_PARTITION_UA自動選擇分區 //$option可選 $topic->produce(RD_KAFKA_PARTITION_UA, 0, "qkl . $i", $option); } $len = $rk->getOutQLen(); while ($len > 0) { $len = $rk->getOutQLen(); var_dump($len); $rk->poll(50); }運行生產者
php producer.php # output int(20) int(20) int(20) int(20) int(0) # 你可以查看你剛才上面啟動的消費者shell應該會輸出消息 qkl . 0 qkl . 1 qkl . 2 qkl . 3 qkl . 4 qkl . 5 qkl . 6 qkl . 7 qkl . 8 qkl . 9 qkl . 10 qkl . 11 qkl . 12 qkl . 13 qkl . 14 qkl . 15 qkl . 16 qkl . 17 qkl . 18 qkl . 19Low Level 消費者
setDrMsgCb(function ($kafka, $message) { file_put_contents("./c_dr_cb.log", var_export($message, true), FILE_APPEND); }); $conf->setErrorCb(function ($kafka, $err, $reason) { file_put_contents("./err_cb.log", sprintf("Kafka error: %s (reason: %s)", rd_kafka_err2str($err), $reason).PHP_EOL, FILE_APPEND); }); //設置消費組 $conf->set("group.id", "myConsumerGroup"); $rk = new RdKafkaConsumer($conf); $rk->addBrokers("127.0.0.1"); $topicConf = new RdKafkaTopicConf(); $topicConf->set("request.required.acks", 1); //在interval.ms的時間內自動提交確認、建議不要啟動 //$topicConf->set("auto.commit.enable", 1); $topicConf->set("auto.commit.enable", 0); $topicConf->set("auto.commit.interval.ms", 100); // 設置offset的存儲為file //$topicConf->set("offset.store.method", "file"); // 設置offset的存儲為broker $topicConf->set("offset.store.method", "broker"); //$topicConf->set("offset.store.path", __DIR__); //smallest:簡單理解為從頭開始消費,其實等價于上面的 earliest //largest:簡單理解為從最新的開始消費,其實等價于上面的 latest //$topicConf->set("auto.offset.reset", "smallest"); $topic = $rk->newTopic("test", $topicConf); // 參數1消費分區0 // RD_KAFKA_OFFSET_BEGINNING 重頭開始消費 // RD_KAFKA_OFFSET_STORED 最后一條消費的offset記錄開始消費 // RD_KAFKA_OFFSET_END 最后一條消費 $topic->consumeStart(0, RD_KAFKA_OFFSET_BEGINNING); //$topic->consumeStart(0, RD_KAFKA_OFFSET_END); // //$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED); while (true) { //參數1表示消費分區,這里是分區0 //參數2表示同步阻塞多久 $message = $topic->consume(0, 12 * 1000); if (is_null($message)) { sleep(1); echo "No more messages "; continue; } switch ($message->err) { case RD_KAFKA_RESP_ERR_NO_ERROR: var_dump($message); break; case RD_KAFKA_RESP_ERR__PARTITION_EOF: echo "No more messages; will wait for more "; break; case RD_KAFKA_RESP_ERR__TIMED_OUT: echo "Timed out "; break; default: throw new Exception($message->errstr(), $message->err); break; } }High LEVEL消費者
assign(); // $kafka->assign([new RdKafkaTopicPartition("qkl01", 0, 0)]); break; case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS: echo "Revoke: "; var_dump($partitions); $kafka->assign(NULL); break; default: throw new Exception($err); } } // Set a rebalance callback to log partition assignments (optional) $conf->setRebalanceCb(function(RdKafkaKafkaConsumer $kafka, $err, array $partitions = null) { rebalance($kafka, $err, $partitions); }); // Configure the group.id. All consumer with the same group.id will consume // different partitions. $conf->set("group.id", "test-110-g100"); // Initial list of Kafka brokers $conf->set("metadata.broker.list", "192.168.216.122"); $topicConf = new RdKafkaTopicConf(); $topicConf->set("request.required.acks", -1); //在interval.ms的時間內自動提交確認、建議不要啟動 $topicConf->set("auto.commit.enable", 0); //$topicConf->set("auto.commit.enable", 0); $topicConf->set("auto.commit.interval.ms", 100); // 設置offset的存儲為file $topicConf->set("offset.store.method", "file"); $topicConf->set("offset.store.path", __DIR__); // 設置offset的存儲為broker // $topicConf->set("offset.store.method", "broker"); // Set where to start consuming messages when there is no initial offset in // offset store or the desired offset is out of range. // "smallest": start from the beginning $topicConf->set("auto.offset.reset", "smallest"); // Set the configuration to use for subscribed/assigned topics $conf->setDefaultTopicConf($topicConf); $consumer = new RdKafkaKafkaConsumer($conf); //$KafkaConsumerTopic = $consumer->newTopic("qkl01", $topicConf); // Subscribe to topic "test" $consumer->subscribe(["qkl01"]); echo "Waiting for partition assignment... (make take some time when "; echo "quickly re-joining the group after leaving it.) "; while (true) { $message = $consumer->consume(120*1000); switch ($message->err) { case RD_KAFKA_RESP_ERR_NO_ERROR: var_dump($message); // $consumer->commit($message); // $KafkaConsumerTopic->offsetStore(0, 20); break; case RD_KAFKA_RESP_ERR__PARTITION_EOF: echo "No more messages; will wait for more "; break; case RD_KAFKA_RESP_ERR__TIMED_OUT: echo "Timed out "; break; default: throw new Exception($message->errstr(), $message->err); break; } }消費組特別說明
特別注意,High LEVEL消費者設置的消費組,kafka服務器才會記錄, Low Level消費者設置的消費組,服務器不會記錄
具體查看消費組信息,你可以翻閱本篇文章
查看服務器元數據(topic/partition/broker)setDrMsgCb(function ($kafka, $message) { file_put_contents("./xx.log", var_export($message, true), FILE_APPEND); }); $conf->setErrorCb(function ($kafka, $err, $reason) { printf("Kafka error: %s (reason: %s) ", rd_kafka_err2str($err), $reason); }); $conf->set("group.id", "myConsumerGroup"); $rk = new RdKafkaConsumer($conf); $rk->addBrokers("127.0.0.1"); $allInfo = $rk->metadata(true, NULL, 60e3); $topics = $allInfo->getTopics(); echo rd_kafka_offset_tail(100); echo "--"; echo count($topics); echo "--"; foreach ($topics as $topic) { $topicName = $topic->getTopic(); if ($topicName == "__consumer_offsets") { continue ; } $partitions = $topic->getPartitions(); foreach ($partitions as $partition) { // $rf = new ReflectionClass(get_class($partition)); // foreach ($rf->getMethods() as $f) { // var_dump($f); // } // die(); $topPartition = new RdKafkaTopicPartition($topicName, $partition->getId()); echo "當前的話題:" . ($topPartition->getTopic()) . " - " . $partition->getId() . " - "; echo "offset:" . ($topPartition->getOffset()) . PHP_EOL; } }如果需遠端生產和消費
vim config/server.properties advertised.listeners=PLAINTEXT://ip:9092 # ip 未你kafka的外網ip即可分享一個打包好的php-rdkafka的類庫
https://github.com/qkl9527/php-rdkafka-class
參考文獻Kafka文檔
文章版權歸作者所有,未經允許請勿轉載,若此文章存在違規行為,您可以聯系管理員刪除。
轉載請注明本文地址:http://specialneedsforspecialkids.com/yun/29117.html
摘要:閱讀本教程前最好先嘗試閱讀下的實踐自帶命令實踐嘗試實踐的知識創建話題生產消息消費消息話題信息獲取消費組獲取消費組的自帶的命令安裝目錄的目錄下代表我們會使用的腳本 閱讀本教程前最好先嘗試閱讀:PHP下kafka的實踐 自帶命令實踐 嘗試實踐的kafka知識: 創建話題 生產消息 消費消息 話題信息 獲取消費組 獲取消費組的offset 自帶的命令 # kafka安裝目錄的bin目錄下...
閱讀 2222·2021-09-24 10:31
閱讀 3875·2021-09-22 15:16
閱讀 3395·2021-09-22 10:02
閱讀 1010·2021-09-22 10:02
閱讀 1822·2021-09-08 09:36
閱讀 1974·2019-08-30 14:18
閱讀 609·2019-08-30 10:51
閱讀 1863·2019-08-29 11:08