《p》Kafka將數(shù)據(jù)持久化到了硬盤(pán)上，允許你配置一定的策略對(duì)數(shù)據(jù)清理，清理的策略有兩個(gè)，刪除和壓縮。 《br》 數(shù)據(jù)清理的方式 《br》 刪除 《br》 log.cleanup.policy=delete啟用刪除策略 《br》 直接刪除，刪除后的消息不可恢復(fù)。可配置以下兩個(gè)策略： 《br》 清理超過(guò)指定時(shí)間清理： 《br》 log.retention.hours=16 《br》 超過(guò)指定大小后，刪除舊的消息： 《br》 log.retention.bytes=1073741824 《br》 為了避免在刪除時(shí)阻塞讀操作，采用了copy-on-write形式的實(shí)現(xiàn)，刪除操作進(jìn)行時(shí)，讀取操作的二分查找功能實(shí)際是在一個(gè)靜態(tài)的快照副本上進(jìn)行的，這類(lèi)似于Java的CopyOnWriteArrayList。 《br》 壓縮 《br》 將數(shù)據(jù)壓縮，只保留每個(gè)key最后一個(gè)版本的數(shù)據(jù)。 《br》 首先在broker的配置中設(shè)置log.cleaner.enable=true啟用cleaner，這個(gè)默認(rèn)是關(guān)閉的。 《br》 在topic的配置中設(shè)置log.cleanup.policy=compact啟用壓縮策略。 《br》 壓縮策略的細(xì)節(jié)《/p》《p》如上圖，在整個(gè)數(shù)據(jù)流中，每個(gè)Key都有可能出現(xiàn)多次，壓縮時(shí)將根據(jù)Key將消息聚合，只保留最后一次出現(xiàn)時(shí)的數(shù)據(jù)。這樣，無(wú)論什么時(shí)候消費(fèi)消息，都能拿到每個(gè)Key的最新版本的數(shù)據(jù)。 《br》 壓縮后的offset可能是不連續(xù)的，比如上圖中沒(méi)有5和7，因?yàn)檫@些offset的消息被merge了，當(dāng)從這些offset消費(fèi)消息時(shí)，將會(huì)拿到比這個(gè)offset大的offset對(duì)應(yīng)的消息，比如，當(dāng)試圖獲取offset為5的消息時(shí)，實(shí)際上會(huì)拿到offset為6的消息，并從這個(gè)位置開(kāi)始消費(fèi)。 《br》 這種策略只適合特俗場(chǎng)景，比如消息的key是用戶ID，消息體是用戶的資料，通過(guò)這種壓縮策略，整個(gè)消息集里就保存了所有用戶最新的資料。 《br》 壓縮策略支持刪除，當(dāng)某個(gè)Key的最新版本的消息沒(méi)有內(nèi)容時(shí)，這個(gè)Key將被刪除，這也符合以上邏輯

　　Kafka定時(shí)清除過(guò)期數(shù)據(jù)2017-09-04 14:04《p》Kafka將數(shù)據(jù)持久化到了硬盤(pán)上，允許你配置一定的策略對(duì)數(shù)據(jù)清理，清理的策略有兩個(gè)，刪除和壓縮。 《br》 數(shù)據(jù)清理的方式 《br》 刪除 《br》 log.cleanup.policy=delete啟用刪除策略 《br》 直接刪除，刪除后的消息不可恢復(fù)。可配置以下兩個(gè)策略： 《br》 清理超過(guò)指定時(shí)間清理： 《br》 log.retention.hours=16 《br》 超過(guò)指定大小后，刪除舊的消息： 《br》 log.retention.bytes=1073741824 《br》 為了避免在刪除時(shí)阻塞讀操作，采用了copy-on-write形式的實(shí)現(xiàn)，刪除操作進(jìn)行時(shí)，讀取操作的二分查找功能實(shí)際是在一個(gè)靜態(tài)的快照副本上進(jìn)行的，這類(lèi)似于Java的CopyOnWriteArrayList。 《br》 壓縮 《br》 將數(shù)據(jù)壓縮，只保留每個(gè)key最后一個(gè)版本的數(shù)據(jù)。 《br》 首先在broker的配置中設(shè)置log.cleaner.enable=true啟用cleaner，這個(gè)默認(rèn)是關(guān)閉的。 《br》 在topic的配置中設(shè)置log.cleanup.policy=compact啟用壓縮策略。 《br》 壓縮策略的細(xì)節(jié)《/p》《p》如上圖，在整個(gè)數(shù)據(jù)流中，每個(gè)Key都有可能出現(xiàn)多次，壓縮時(shí)將根據(jù)Key將消息聚合，只保留最后一次出現(xiàn)時(shí)的數(shù)據(jù)。這樣，無(wú)論什么時(shí)候消費(fèi)消息，都能拿到每個(gè)Key的最新版本的數(shù)據(jù)。 《br》 壓縮后的offset可能是不連續(xù)的，比如上圖中沒(méi)有5和7，因?yàn)檫@些offset的消息被merge了，當(dāng)從這些offset消費(fèi)消息時(shí)，將會(huì)拿到比這個(gè)offset大的offset對(duì)應(yīng)的消息，比如，當(dāng)試圖獲取offset為5的消息時(shí)，實(shí)際上會(huì)拿到offset為6的消息，并從這個(gè)位置開(kāi)始消費(fèi)。 《br》 這種策略只適合特俗場(chǎng)景，比如消息的key是用戶ID，消息體是用戶的資料，通過(guò)這種壓縮策略，整個(gè)消息集里就保存了所有用戶最新的資料。 《br》 壓縮策略支持刪除，當(dāng)某個(gè)Key的最新版本的消息沒(méi)有內(nèi)容時(shí)，這個(gè)Key將被刪除，這也符合以上邏輯

相關(guān)電子資料下載

• kafka相關(guān)命令詳解 81
• Kafka架構(gòu)技術(shù)：Kafka的架構(gòu)和客戶端API設(shè)計(jì) 80
• Spring Kafka的各種用法 91
• kafka client在 spring如何實(shí)現(xiàn) 104
• 降本增效：Grab如何在云上將Kafka消費(fèi)者流量成本降到零 214
• 監(jiān)控Kafka集群的常用的方法和工具介紹 167
• 從Kafka中學(xué)習(xí)高性能系統(tǒng)如何設(shè)計(jì) 211
• 物通博聯(lián)WG783-5G-kafka（帶kafka協(xié)議開(kāi)發(fā)調(diào)試）系列工業(yè)智能網(wǎng)關(guān) 134
• 物通博聯(lián)5G-kafka工業(yè)智能網(wǎng)關(guān)輕松實(shí)現(xiàn)數(shù)據(jù)采集kafka上云 155
• 物通博聯(lián)5G-kafka工業(yè)網(wǎng)關(guān)實(shí)現(xiàn)kafka協(xié)議對(duì)接到云平臺(tái) 132