kafka在java代碼中常用方法

『壹』 Java使用kafka發送消息沒有生效

一般消息發不出去很大可能都是配置或環境的問題
1、排查環境是否有問題，zookeeper節點是否存活，kafka節點是否存活，通過命令行的方式能否發出去消息（使用kafka-console-procer.sh）,如果通過命令行都發不出去那就是集群的問題了。
2、網路問題，調用機器和集群之間網路是否通暢
3、調用時配置的host、port和集群中配置的是否一致，是否需要使用主機名而不是ip
4、客戶端api版本是否和服務端差別太大導致不兼容
5、防火牆問題，關閉集群的防火牆實時
諸如此類，可能性太多就不一 一列舉了。
你這既然有列印堆棧，如果報錯肯定有異常信息的，可能卡住的時間比較長，耐心等待吧，祝你早日解決bug。

『貳』 如何通過java實現kafka集群的配額

序列化就是一種用來處理對象流的機制，所謂對象流也就是將對象的內容進行流化。可以對流化後的對象進行讀寫操作，也可將流化後的對象傳輸於網路之間。序列化是為了解決在對對象流進行讀寫操作時所引發的問題。
序列化的實現：將需要被序列化的類實現Serializable介面，該介面沒有需要實現的方法，implements Serializable只是為了標注該對象是可被序列化的，然後使用一個輸出流(如：FileOutputStream)來構造一個ObjectOutputStream(對象流)對象，接著，使用ObjectOutputStream對象的writeObject(Object obj)方法就可以將參數為obj的對象寫出(即保存其狀態)，要恢復的話則用輸入流。

『叄』 kafka簡介

一、kafka定義

二、kafka的優勢

三、kafka的原理

四、kafka起源

一、Kafka是最初由Linkedin公司開發，是一個分布式、支持分區的（partition）、多副本的（replica），基於zookeeper協調的分布式消息系統，它的最大的特性就是可以實時的處理大量數據以滿足各種需求場景：比如基於hadoop的批處理系統、低延遲的實時系統、storm/Spark流式處理引擎，web/nginx日誌、訪問日誌，消息服務等等，用scala語言編寫，Linkedin於2010年貢獻給了Apache基金會並成為頂級開源項目。

二、kafka的優勢

高吞吐量、低延遲：kafka美妙之處是可以處理幾十萬條信息，它的延遲最低只有幾毫秒，每個topic可以分多個partition，consumer

group對partition進行consume操作。

可擴展性：kafka集群支持熱擴展

持久化、可靠性：消息被持久化到本地磁碟，並且支持數據備份防止數據丟失

容錯性：允許集群中節點失敗（若副本數量為n，則允許n-1個節點失敗）

高並發：支持數千個客戶端同時讀寫

三、kafka的原理

kafka是如何實現以上所述這幾點，我們逐一說明：

1.高吞吐量、低延遲

kafka在設計之初就是為了針對大數據量的傳輸處理，高吞吐量、低延遲最主要看的就是單位時間內所能讀寫的數據總量，我們先來看生產端。

kafka採取了一定量的批處理機制，即當生產數據達到一定數量或者達到時間窗口後，將所收集到的數據一批次的提交到伺服器，我們假設處理一次數據的時間為1ms，那每秒鍾能處理1000條，延時為1ms，如果此時將處理間隔變成9ms，即每10ms處理一批數據，假設這段時間接收到100條處理，那每秒則能處理10000條，但是延時變成了10ms。為了獲得最大的吞吐量，需要犧牲一定的延遲，但是這樣的犧牲是值得的。當確定了這種小批量方式之後，高速的寫則取決於kafka自身寫磁碟的速度了。而由於kafka本身對數據不做任何的處理，只管寫入數據，保管數據，分發數據，因此會是一種批量順序寫入數據的情況，而磁碟的讀寫速度大量消耗在定址上，也就是隨機讀寫，但是對於順序寫入的速度是非常快的，甚至能媲美內存的隨機寫入速度。有人做過一個對比，普通磁碟順序寫入每秒能達到53.2M/s，SSD的順序寫入速度為42.2M/s，內存的順序寫入速度為358.2M/s。kafka正是利用了這個特性，順序寫入，速度相對較快。而kafka本身雖然也是寫入磁碟持久化數據，但實際上kafka是將數據順序寫入頁緩存中（page cache），然後由操作系統自行決定何時寫到磁碟上，因此kafka的寫操作能在每秒輕輕鬆鬆達到寫入數十萬條記錄。並且基於kafka的動態擴展，這個數字還能不斷增大。

kafka在消費端也有著高吞吐量，由於kafka是將數據寫入到頁緩存中，同時由於讀寫相間的間隔並不大，很大可能性會在緩存中命中，從而保證高吞吐量。另外kafka由於本身不對數據做任何的修改，完全使用零拷貝技術，大大提升數據的讀取能力。

2.kafka每個節點叫做broker，而每一個broker都是獨立運行的，可以隨時加入kafka集群，集群的心跳管理是由zookeeper負責，新加入的broker只要broker id不與原有的沖突就能順利的加入集群中，實現動態擴展。

3.kafka的持久化在上面已經提到，kafka繞過了java的堆處理數據，直接將數據寫入頁緩存，然後由操作系統來管理頁緩存寫入磁碟，實現持久化。kafka每一個主題topic是一個業務數據，他可由多個partition組成，而每個partition可以有多個replica副本，用於保證數據的可靠性。replica分為兩個角色，一個是leader，一個是追隨者，同一時間，每一個partition只能有一個leader，其他都是追問隨者，laeder負責接收數據並寫入log，而追隨者不能被用戶寫入數據，只是從leader角色的replica副本中同步log寫入自己的log，保持數據同步。kafka中有一個概念，ISR，全稱是in-sync

replica，即所有可用的replica副本，這里的ISR數量只要大於1，這個partition就能正常運作，因此容錯性非常好，假設n個replica，那最多可以壞n-1個replica的情況下，還能保持系統正常運行。當replica遲滯到一定時間後，會被kafka從ISR中剔除，當再次同步後，可以再次加入ISR，如果這時候leader出現問題，會從ISR中重新選舉一個leader，原先的leader再次同步成功後會重新加入ISR，成為一個flower。

4.上面提到了kafka的ISR機制，kafka的容錯性就是由ISR的機制來保證的。

5.kafka集群可以動態擴展broker，多個partition同時寫入消費數據，實現真正的高並發。

四、kafka的起源

kafka起源於LinkedIn公司，當時領英公司需要收集兩大類數據，一是業務系統和應用程序的性能監控指標數據，而是用戶的操作行為數據。當時為了收集這兩類數據，領英自研了兩套相應的數據收集系統，但是這兩套系統都存在一些弊端，無法實現實時交互、實時性差、維護成本高。因此領英的工程師希望找到一個統一的組件來收集分發消費這些大批量的數據，ActiveMQ由於擴展性不足，不能支撐大數據量而被拋棄，從而決定自研一套滿足需求的系統組件，也就是kafka。

kafka的設計之初主要有三個目標：

1.為生產者和消費者提供一套簡單的API

2.降低網路傳輸和磁碟存儲開銷

3.具有高伸縮性架構

目前kafka可以算是超額完成了目標。

kafka的名稱由來也很有意思，因為kafka系統的寫操作性能特別強，因此想使用一個作家的名字來命名kafka，而Jay Kreps，kafka的三位作者之一，在上大學的時候很喜歡Franz Kafka，因此起來這樣一個名字。

kafka在2010年開源，2011年7月正式進入Apache進行孵化，2012年10月順利畢業，後成為Apache的頂級項目。

『肆』 java kafka 怎麼傳輸對象

1、zookeeper集群 搭建在110， 111,112
2、kafka使用3個節點110， 111,112 修改配置文件config/server.properties broker.id=110 host.name=192.168.1.110 log.dirs=/usr/local/kafka_2.10-0.8.2.0/logs 復制到其他兩個節點，然後修改對應節點上的config/server.pro
3、啟動,在三個節點分別執行 bin/kafka-server-start.sh config/server.properties >/dev/null 2>&1 &
4、 創建主題 bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 3 --topic test
5 、查看主題詳細 bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test --topic test Topic:test PartitionCount:3 R...

『伍』 kafka低版本的怎麼用java查詢給定broker上所有的日誌目錄信息

1. 日誌存儲格式

最新版本的kafka日誌是以批為單位進行日誌存儲的，所謂的批指的是kafka會將多條日誌壓縮到同一個batch中，然後以batch為單位進行後續的諸如索引的創建和消息的查詢等工作。

對於每個批次而言，其默認大小為4KB，並且保存了整個批次的起始位移和時間戳等元數據信息，而對於每條消息而言，其位移和時間戳等元數據存儲的則是相對於整個批次的元數據的增量，通過這種方式，kafka能夠減少每條消息中數據佔用的磁碟空間。

這里我們首先展示一下每個批次的數據格式：

圖中K1的數據有V1、V3和V4，經過壓縮之後只有V4保留了下來，K2的數據則有V2、V6和V10，壓縮之後也只有V10保留了下來；同理可推斷其他的Key的數據。

另外需要注意的是，kafka開啟日誌壓縮使用的是log.cleanup.policy，其默認值為delete，也即我們正常使用的策略，如果將其設置為compaction，則開啟了日誌壓縮策略，但是需要注意的是，開啟了日誌壓縮策略並不代表kafka會清理歷史數據，只有將log.cleaner.enable設置為true才會定時清理歷史數據。

在kafka中，其本身也在使用日誌壓縮策略，主要體現在kafka消息的偏移量存儲。在舊版本中，kafka將每個consumer分組當前消費的偏移量信息保存在zookeeper中，但是由於zookeeper是一款分布式協調工具，其對於讀操作具有非常高的性能，但是對於寫操作性能比較低，而consumer的位移提交動作是非常頻繁的，這勢必會導致zookeeper成為kafka消息消費的瓶頸。

因而在最新版本中，kafka將分組消費的位移數據存儲在了一個特殊的topic中，即__consumer_offsets，由於每個分組group的位移信息都會提交到該topic，因而kafka默認為其設置了非常多的分區，也即50個分區。

另外，consumer在提交位移時，使用的key為groupId+topic+partition，而值則為當前提交的位移，也就是說，對於每一個分組所消費的topic的partition，其都只會保留最新的位移。如果consumer需要讀取位移，那麼只需要按照上述格式組裝key，然後在該topic中讀取最新的消息數據即可。

『陸』 kafka怎麼發布訂閱 怎麼在java中實現

這是我們項目中用到的代碼

publicclassProcerService{
	privatestaticLoggerlog=Logger.getLogger(ProcerService.class);
	privatestaticProcer<String,String>procer=null;
	privatestaticStringserviceIp=PropertiesUtils.getValue("/epoo.properties","bootstrap.servers");
	=PropertiesUtils.getValue("/epoo.properties","name");
publicbooleaninitProcer(){
Propertiesprops=newProperties();


//dataPlace.getIp()
props.put("bootstrap.servers",serviceIp);
props.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
props.put("acks","-1");
procer=newKafkaProcer(props);
try{
List<PartitionInfo>list=procer.partitionsFor(serviceName);

}catch(Exceptione){
JOptionPane.showMessageDialog(null,e.getMessage(),"錯誤",JOptionPane.YES_OPTION);
log.error(e.getMessage());
returnfalse;
}
returntrue;
}

publicvoidsendData(Stringmess){
if(procer==null){
initProcer();
}

procer.send(newProcerRecord<String,String>(serviceName,mess),newCallback(){
@Override
publicvoidonCompletion(RecordMetadatarm,Exceptione){
if(e!=null){
e.printStackTrace();
log.error(e.getMessage());
}
System.out.println("發送到伺服器的Offset:"+rm.offset()+"-----Topic:"+rm.topic()+"-----partition:"+rm.partition());
}
});
}

publicvoidclose(){
if(procer!=null){
procer.close();
}
}
}

『柒』 3分鍾帶你徹底搞懂 Kafka

Kafka到底是個啥？用來幹嘛的？

官方定義如下：

翻譯過來，大致的意思就是，這是一個實時數據處理系統，可以橫向擴展，並高可靠！

實時數據處理 ，從名字上看，很好理解，就是將數據進行實時處理，在現在流行的微服務開發中，最常用實時數據處理平台有 RabbitMQ、RocketMQ 等消息中間件。

這些中間件，最大的特點主要有兩個：

在早期的 web 應用程序開發中，當請求量突然上來了時候，我們會將要處理的數據推送到一個隊列通道中，然後另起一個線程來不斷輪訓拉取隊列中的數據，從而加快程序的運行效率。

但是隨著請求量不斷的增大，並且隊列通道的數據一致處於高負載，在這種情況下，應用程序的內存佔用率會非常高，稍有不慎，會出現內存不足，造成程序內存溢出，從而導致服務不可用。

隨著業務量的不斷擴張，在一個應用程序內，使用這種模式已然無法滿足需求，因此之後，就誕生了各種消息中間件，例如 ActiveMQ、RabbitMQ、RocketMQ等中間件。

採用這種模型，本質就是將要推送的數據，不在存放在當前應用程序的內存中，而是將數據存放到另一個專門負責數據處理的應用程序中，從而實現服務解耦。

消息中間件 ：主要的職責就是保證能接受到消息，並將消息存儲到磁碟，即使其他服務都掛了，數據也不會丟失，同時還可以對數據消費情況做好監控工作。

應用程序 ：只需要將消息推送到消息中間件，然後啟用一個線程來不斷從消息中間件中拉取數據，進行消費確認即可！

引入消息中間件之後，整個服務開發會變得更加簡單，各負其責。

Kafka 本質其實也是消息中間件的一種，Kafka 出自於 LinkedIn 公司，與 2010 年開源到 github。

LinkedIn 的開發團隊，為了解決數據管道問題，起初採用了 ActiveMQ 來進行數據交換，大約是在 2010 年前後，那時的 ActiveMQ 還遠遠無法滿足 LinkedIn 對數據傳遞系統的要求，經常由於各種缺陷而導致消息阻塞或者服務無法正常訪問，為了能夠解決這個問題，LinkedIn 決定研發自己的消息傳遞系統， Kafka 由此誕生 。

在 LinkedIn 公司，Kafka 可以有效地處理每天數十億條消息的指標和用戶活動跟蹤，其強大的處理能力，已經被業界所認可，並成為大數據流水線的首選技術。

先來看一張圖， 下面這張圖就是 kafka 生產與消費的核心架構模型 ！

如果你看不懂這些概念沒關系，我會帶著大家一起梳理一遍！

簡而言之，kafka 本質就是一個消息系統，與大多數的消息系統一樣，主要的特點如下：

與 ActiveMQ、RabbitMQ、RocketMQ 不同的地方在於，它有一個**分區 Partition **的概念。

這個分區的意思就是說，如果你創建的 topic 有5個分區，當你一次性向 kafka 中推 1000 條數據時，這 1000 條數據默認會分配到 5 個分區中，其中每個分區存儲 200 條數據。

這樣做的目的，就是方便消費者從不同的分區拉取數據，假如你啟動 5 個線程同時拉取數據，每個線程拉取一個分區，消費速度會非常非常快！

這是 kafka 與其他的消息系統最大的不同！

和其他的中間件一樣，kafka 每次發送數據都是向 Leader 分區發送數據，並順序寫入到磁碟，然後 Leader 分區會將數據同步到各個從分區 Follower ，即使主分區掛了，也不會影響服務的正常運行。

那 kafka 是如何將數據寫入到對應的分區呢？kafka中有以下幾個原則：

與生產者一樣，消費者主動的去kafka集群拉取消息時，也是從 Leader 分區去拉取數據。

這里我們需要重點了解一個名詞： 消費組 ！

考慮到多個消費者的場景，kafka 在設計的時候，可以由多個消費者組成一個消費組，同一個消費組者的消費者可以消費同一個 topic 下不同分區的數據，同一個分區只會被一個消費組內的某個消費者所消費，防止出現重復消費的問題！

但是不同的組，可以消費同一個分區的數據！

你可以這樣理解，一個消費組就是一個客戶端，一個客戶端可以由很多個消費者組成，以便加快消息的消費能力。

但是，如果一個組下的消費者數量大於分區數量，就會出現很多的消費者閑置。

如果分區數量大於一個組下的消費者數量，會出現一個消費者負責多個分區的消費，會出現消費性能不均衡的情況。

因此，在實際的應用中，建議消費者組的 consumer 的數量與 partition 的數量保持一致！

光說理論可沒用，下面我們就以 centos7 為例，介紹一下 kafka 的安裝和使用。

kafka 需要 zookeeper 來保存服務實例的元信息，因此在安裝 kafka 之前，我們需要先安裝 zookeeper。

zookeeper 安裝環境依賴於 jdk，因此我們需要事先安裝 jdk

下載zookeeper，並解壓文件包

創建數據、日誌目錄

配置zookeeper

重新配置 dataDir 和 dataLogDir 的存儲路徑

最後，啟動 Zookeeper 服務

到官網 http://kafka.apache.org/downloads.html 下載想要的版本，我這里下載是最新穩定版 2.8.0 。

按需修改配置文件 server.properties （可選）

server.properties 文件內容如下：

其中有四個重要的參數：

可根據自己需求修改對應的配置！

啟動 kafka 服務

創建一個名為 testTopic 的主題，它只包含一個分區，只有一個副本：

運行 list topic 命令，可以看到該主題。

輸出內容：

Kafka 附帶一個命令行客戶端，它將從文件或標准輸入中獲取輸入，並將其作為消息發送到 Kafka 集群。默認情況下，每行將作為單獨的消息發送。

運行生產者，然後在控制台中鍵入一些消息以發送到伺服器。

輸入兩條內容並回車：

Kafka 還有一個命令行使用者，它會將消息轉儲到標准輸出。

輸出結果如下：

本文主要圍繞 kafka 的架構模型和安裝環境做了一些初步的介紹，難免會有理解不對的地方，歡迎網友批評、吐槽。

由於篇幅原因，會在下期文章中詳細介紹 java 環境下 kafka 應用場景！

『捌』 如何寫java程序代碼測試kafka

我這里是使用的是，kafka自帶的zookeeper。
以及關於kafka的日誌文件啊，都放在默認里即/tmp下，我沒修改。保存默認的

1、 [hadoop@sparksinglenode kafka_2.10-0.8.1.1]$ jps
2625 Jps
2、 [hadoop@sparksinglenode kafka_2.10-0.8.1.1]$ bin/zookeeper-server-start.sh config/zookeeper.properties
此刻，這時，會一直停在這，因為是前端運行。
另開一窗口，
3、 [hadoop@sparksinglenode kafka_2.10-0.8.1.1]$ bin/kafka-server-start.sh config/server.properties
也是前端運行。

『玖』 java客戶端使用kafka時什麼情況下使用kafka client和spring kafka

spring-kafka 是基於 java版的 kafka client與spring的集成，提供了 KafkaTemplate，封裝了各種方法，方便操作
所以你使用spring的情況下，可以用spring-kafka，當然直接用kafka client也行

『拾』 Kafka-概述

    Kafka是最初由Linkedin公司開發，是一個分布式、支持分區的(partition)、多副本的(replica)，基於zookeeper協調的分布式消息系統，它的最大的特性就是可以實時的處理大量數據以滿足各種需求場景：比如基於hadoop的批處理系統、低延遲的實時系統、storm/Spark流式處理引擎，web/nginx日誌、訪問日誌，消息服務等等，用scala語言編寫，Linkedin於2010年貢獻給了Apache基金會並成為頂級開源 項目。

JMS(Java Message Service)是Java提供的一套技術規范

    用來異構系統 集成通信，緩解系統瓶頸，提高系統的伸縮性增強系統用戶體驗，使得系統模塊化和組件化變得可行並更加靈活

(1) 點對點模式(一對一，消費者主動拉取數據，消息收到後消息清除)

    點對點模型通常是一個基於拉取或者輪詢的消息傳送模型，這種模型從隊列中請求信息，而不是將消息推送到客戶端。這個模型的特點是發送到隊列的消息被一個且只有一個接收者接收處理，即使有多個消息監聽者也是如此。

(2) 發布/訂閱模式(一對多，數據生產後，推送給所有訂閱者)
    發布訂閱模型則是一個基於推送的消息傳送模型。發布訂閱模型可以有多種不同的訂閱者，臨時訂閱者只在主動監聽主題時才接收消息，而持久訂閱者則監聽主題的所有消息，即使當前訂閱者不可用，處於離線狀態。

    kafka每秒可以處理幾十萬條消息，它的延遲最低只有幾毫秒，每個topic可以分多個partition, consumer group 對partition進行consume操作。

kafka集群支持熱擴展

消息被持久化到本地磁碟，並且支持數據備份防止數據丟失

允許集群中節點失敗（若副本數量為n,則允許n-1個節點失敗）

支持數千個客戶端同時讀寫

    一個公司可以用Kafka可以收集各種服務的log，通過kafka以統一介面服務的方式開放給各種consumer，例如hadoop、Hbase、Solr等。

    解耦和生產者和消費者、緩存消息等。

    Kafka經常被用來記錄web用戶或者app用戶的各種活動，如瀏覽網頁、搜索、點擊等活動，這些活動信息被各個伺服器發布到kafka的topic中，然後訂閱者通過訂閱這些topic來做實時的監控分析，或者裝載到hadoop、數據倉庫中做離線分析和挖掘。

    Kafka也經常用來記錄運營監控數據。包括收集各種分布式應用的數據，生產各種操作的集中反饋，比如報警和報告。

    比如spark streaming和storm

    Kafka每個主題的多個分區日誌分布式地存儲在Kafka集群上，同時為了故障容錯，每個分區都會以副本的方式復制到多個消息代理節點上。其中一個節點會作為主副本(Leader)，其他節點作為備份副本(Follower，也叫作從副本)。主副本會負責所有的客戶端讀寫操作，備份副本僅僅從主副本同步數據。當主副本出現故障時，備份副本中的一個副本會被選擇為新的主副本。因為每個分區的副本中只有主副本接受讀寫，所以每個伺服器端都會作為某些分區的主副本，以及另外一些分區的備份副本，這樣Kafka集群的所有服務端整體上對客戶端是負載均衡的。

    Kafka的生產者和消費者相對於伺服器端而言都是客戶端。

    Kafka生產者客戶端發布消息到服務端的指定主題，會指定消息所屬的分區。生產者發布消息時根據消息是否有鍵，採用不同的分區策略。消息沒有鍵時，通過輪詢方式進行客戶端負載均衡；消息有鍵時，根據分區語義（例如hash）確保相同鍵的消息總是發送到同一分區。

    Kafka的消費者通過訂閱主題來消費消息，並且每個消費者都會設置一個消費組名稱。因為生產者發布到主題的每一條消息都只會發送給消費者組的一個消費者。所以，如果要實現傳統消息系統的「隊列」模型，可以讓每個消費者都擁有相同的消費組名稱，這樣消息就會負責均衡到所有的消費者；如果要實現「發布-訂閱」模型，則每個消費者的消費者組名稱都不相同，這樣每條消息就會廣播給所有的消費者。

    分區是消費者現場模型的最小並行單位。如下圖（圖1）所示，生產者發布消息到一台伺服器的3個分區時，只有一個消費者消費所有的3個分區。在下圖（圖2）中，3個分區分布在3台伺服器上，同時有3個消費者分別消費不同的分區。假設每個伺服器的吞吐量時300MB，在下圖（圖1）中分攤到每個分區只有100MB，而在下圖（圖2）中，集群整體的吞吐量有900MB。可以看到，增加伺服器節點會提升集群的性能，增加消費者數量會提升處理性能。

    同一個消費組下多個消費者互相協調消費工作，Kafka會將所有的分區平均地分配給所有的消費者實例，這樣每個消費者都可以分配到數量均等的分區。Kafka的消費組管理協議會動態地維護消費組的成員列表，當一個新消費者加入消費者組，或者有消費者離開消費組，都會觸發再平衡操作。

    Kafka的消費者消費消息時，只保證在一個分區內的消息的完全有序性，並不保證同一個主題匯中多個分區的消息順序。而且，消費者讀取一個分區消息的順序和生產者寫入到這個分區的順序是一致的。比如，生產者寫入「hello」和「Kafka」兩條消息到分區P1，則消費者讀取到的順序也一定是「hello」和「Kafka」。如果業務上需要保證所有消息完全一致，只能通過設置一個分區完成，但這種做法的缺點是最多隻能有一個消費者進行消費。一般來說，只需要保證每個分區的有序性，再對消息假設鍵來保證相同鍵的所有消息落入同一分區，就可以滿足絕大多數的應用。