1、消費者代碼
用到消費者，是以也必須先把前面寫過的生産者代碼也貼一下吧
生産者代碼與自定義partition
使用maven導包
<dependencies>
<dependency>
<groupId>com.alibaba.jstorm</groupId>
<artifactId>jstorm-core</artifactId>
<version>2.1.1</version>
<exclusions>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-nop</artifactId>
</exclusion>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-jdk14</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.8.2</artifactId>
<version>0.8.1</version>
<exclusions>
<exclusion>
<artifactId>jmxtools</artifactId>
<groupId>com.sun.jdmk</groupId>
</exclusion>
<exclusion>
<artifactId>jmxri</artifactId>
<groupId>com.sun.jmx</groupId>
</exclusion>
<exclusion>
<artifactId>jms</artifactId>
<groupId>javax.jms</groupId>
</exclusion>
<exclusion>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
</exclusion>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</exclusion>
</exclusions>
</dependency>
</dependencies>
/**
* 這是一個簡單的Kafka producer代碼
* 包含兩個功能:
* 1、資料發送
* 2、資料按照自定義的partition政策進行發送
*
*
* KafkaSpout的類
*/
public class KafkaProducerSimple {
public static void main(String[] args) {
/**
* 1、指定目前kafka producer生産的資料的目的地
* 建立topic可以輸入以下指令，在kafka叢集的任一節點進行建立。
* bin/kafka-topics.sh --create --zookeeper mini1:2181 --replication-factor 2 --partitions 3 --topic test
*/
String TOPIC = "orderMq";
/**
* 2、讀取配置檔案
*/
Properties props = new Properties();
/*
* key.serializer.class預設為serializer.class key的序列化使用哪個類
*/
props.put("serializer.class", "kafka.serializer.StringEncoder");
/*
* kafka broker對應的主機，格式為host1:port1,host2:port2
*/
props.put("metadata.broker.list", "mini1:9092,mini2:9092,mini3:9092");
/*
* request.required.acks,設定發送資料是否需要服務端的回報,有三個值0,1,-1
* 0，意味着producer永遠不會等待一個來自broker的ack，這就是0.7版本的行為。
* 這個選項提供了最低的延遲，但是持久化的保證是最弱的，當server挂掉的時候會丢失一些資料。
* 1，意味着在leader replica已經接收到資料後，producer會得到一個ack。
* 這個選項提供了更好的持久性，因為在server确認請求成功處理後，client才會傳回。
* 如果剛寫到leader上，還沒來得及複制leader就挂了，那麼消息才可能會丢失。
* -1，意味着在所有的ISR都接收到資料後，producer才得到一個ack。
* 這個選項提供了最好的持久性，隻要還有一個replica存活，那麼資料就不會丢失
*/
props.put("request.required.acks", "1");
/*
* 可選配置，如果不配置，則使用預設的partitioner partitioner.class
* 預設值：kafka.producer.DefaultPartitioner
* 用來把消息分到各個partition中，預設行為是對key進行hash。
*/
props.put("partitioner.class", "com.scu.kafka.MyLogPartitioner");
// props.put("partitioner.class", "kafka.producer.DefaultPartitioner");
/**
* 3、通過配置檔案，建立生産者
*/
Producer<String, String> producer = new Producer<String, String>(new ProducerConfig(props));
/**
* 4、通過for循環生産資料
*/
for (int messageNo = ; messageNo < ; messageNo++) {
/**
* 5、調用producer的send方法發送資料
* 注意：這裡需要指定 partitionKey，用來配合自定義的MyLogPartitioner進行資料分發
*/
producer.send(new KeyedMessage<String, String>(TOPIC, messageNo + "", "appid" + UUID.randomUUID() + "itcast"));
}
}
}
public class MyLogPartitioner implements Partitioner {
private static Logger logger = Logger.getLogger(MyLogPartitioner.class);
public MyLogPartitioner(VerifiableProperties props) {
}
/**
*
* @param obj 傳來的key 用它來進行hash分到partition
* @param numPartitions 幾個partition 如果叢集中已存在該topic，那麼partition數為原本存在數，否則預設是2
* @return 生産到哪個partition
*/
public int partition(Object obj, int numPartitions) {
return Integer.parseInt(obj.toString())%numPartitions;
}
}
注：orderMq這個topic很早就通過指令行建立好了，指定了partition是3個。
下面是消費者代碼
public class KafkaConsumerSimple implements Runnable {
public String title;
public KafkaStream<byte[], byte[]> stream;
public KafkaConsumerSimple(String title, KafkaStream<byte[], byte[]> stream) {
this.title = title;
this.stream = stream;
}
public void run() {
System.out.println("開始運作 " + title);
ConsumerIterator<byte[], byte[]> it = stream.iterator();
/**
* 不停地從stream讀取新到來的消息，在等待新的消息時，hasNext()會阻塞
* 如果調用 `ConsumerConnector#shutdown`，那麼`hasNext`會傳回false
* */
while (it.hasNext()) {
MessageAndMetadata<byte[], byte[]> data = it.next();
String topic = data.topic();
int partition = data.partition();
long offset = data.offset();
String msg = new String(data.message());
System.out.println(String.format(
"Consumer: [%s], Topic: [%s], PartitionId: [%d], Offset: [%d], msg: [%s]",
title, topic, partition, offset, msg));
}
System.out.println(String.format("Consumer: [%s] exiting ...", title));
}
public static void main(String[] args) throws Exception{
Properties props = new Properties();
props.put("group.id", "dashujujiagoushi");//消費組組組名，任意取
props.put("zookeeper.connect", "mini1:2181,mini2:2181,mini3:2181");//zookeeper連接配接
props.put("auto.offset.reset", "largest");//最新位置開始消費
props.put("auto.commit.interval.ms", "1000");
props.put("partition.assignment.strategy", "roundrobin");//分區配置設定政策
ConsumerConfig config = new ConsumerConfig(props);
String topic1 = "orderMq";
String topic2 = "paymentMq";
//隻要ConsumerConnector還在的話，consumer會一直等待新消息，不會自己退出
ConsumerConnector consumerConn = Consumer.createJavaConsumerConnector(config);
//定義一個map
Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
topicCountMap.put(topic1, );
//Map<String, List<KafkaStream<byte[], byte[]>> 中String是topic， List<KafkaStream<byte[], byte[]>是對應的流
Map<String, List<KafkaStream<byte[], byte[]>>> topicStreamsMap = consumerConn.createMessageStreams(topicCountMap);
//取出 `kafkaTest` 對應的 streams
List<KafkaStream<byte[], byte[]>> streams = topicStreamsMap.get(topic1);
//建立一個容量為4的線程池
ExecutorService executor = Executors.newFixedThreadPool();
//建立20個consumer threads
for (int i = ; i < streams.size(); i++)
executor.execute(new KafkaConsumerSimple("消費者" + (i + ), streams.get(i)));
}
}
測試：
先執行消費者程式，盡管partition目錄裡面的segment檔案是有以前生成的資料，但是不會列印出來而是一直提示（已經标記為消費狀态的就不再消費了，預設情況就是這樣，可以自己設定從0開始消費）
:: [main-SendThread(mini1:)] DEBUG org.apache.zookeeper.ClientCnxn - Got ping response for sessionid: after ms
:: [main-SendThread(mini1:)] DEBUG org.apache.zookeeper.ClientCnxn - Got ping response for sessionid: after ms
需要進行生産，是以再執行生産者程式，控制台列印如下：
...
Consumer: [消費者1], Topic: [orderMq], PartitionId: [1], Offset: [17857], msg: [appidc977abb2-f0bc-41da-9daa-6b080321947bitcast]
Consumer: [消費者2], Topic: [orderMq], PartitionId: [0], Offset: [17724], msg: [appid9101368e-ac81-4bbf-b2b5-8f2facd41f54itcast]
Consumer: [消費者1], Topic: [orderMq], PartitionId: [1], Offset: [17858], msg: [appidb145da08-bb61-42e7-b140-9fed576c2faeitcast]
Consumer: [消費者1], Topic: [orderMq], PartitionId: [1], Offset: [17859], msg: [appid909a90ae-c0fb-42ac-97de-6d7438895e07itcast]
Consumer: [消費者3], Topic: [orderMq], PartitionId: [2], Offset: [17713], msg: [appid157754b5-6958-4286-9c25-ff67ccc61a42itcast]
Consumer: [消費者3], Topic: [orderMq], PartitionId: [2], Offset: [17714], msg: [appidb93b9355-4713-4e22-823a-756b4fe75bdfitcast]
Consumer: [消費者3], Topic: [orderMq], PartitionId: [2], Offset: [17715], msg: [appidf82ca658-528a-4f40-a023-8a155c15eaa1itcast]
...
精簡下如下
Consumer: [消費者1], Topic: [orderMq], PartitionId: [1], Offset: [17857], msg: [appidc977abb2-f0bc-41da-9daa-6b080321947bitcast]
Consumer: [消費者2], Topic: [orderMq], PartitionId: [0], Offset: [17724], msg: [appid9101368e-ac81-4bbf-b2b5-8f2facd41f54itcast]
Consumer: [消費者3], Topic: [orderMq], PartitionId: [2], Offset: [17713], msg: [appid157754b5-6958-4286-9c25-ff67ccc61a42itcast]
能看到三個消費者對應消費的partition。
那麼考慮以下問題
在建立orderMq的時候指定partition是3個，那麼如果此時我指定建立5個KafkaStream，那麼會怎麼消費呢？
消費者代碼修改兩次如下
topicCountMap.put(topic1, );
ExecutorService executor = Executors.newFixedThreadPool();
再次同上一樣執行，輸出結果能看到隻有3個消費者，是以指定KafkaStream比partition多是沒用的，隻會有對應數量的消費者去消費對應的partition上的資料。
Consumer: [消費者2], Topic: [orderMq], PartitionId: [2], Offset: [26420], msg: [appid4b778b51-33c7-42de-83c2-5b85f8f2428aitcast]
Consumer: [消費者3], Topic: [orderMq], PartitionId: [0], Offset: [26423], msg: [appid86045c25-7b3f-4c82-ad2a-3e8e11958b28itcast]
Consumer: [消費者4], Topic: [orderMq], PartitionId: [1], Offset: [26562], msg: [appid213b5a91-a7bf-4a39-b585-456d95748566itcast]
如果指定的KafkaStream隻有2呢？不做測試了，結果是其中一個消費者會消費2個partition，另外一個消費1個partition中的資料。
生産者，消費者配置檔案解釋
用java api不管是寫生産者代碼還是消費者代碼都使用配置檔案，那麼下面列出了生産者和消費者配置檔案介紹
生産者配置檔案解釋
#指定kafka節點清單，用于擷取metadata，不必全部指定
metadata.broker.list=kafka01:,kafka02:
# 指定分區處理類。預設kafka.producer.DefaultPartitioner，表通過key哈希到對應分區
#partitioner.class=kafka.producer.DefaultPartitioner
# 是否壓縮，預設0表示不壓縮，1表示用gzip壓縮，2表示用snappy壓縮。壓縮後消息中會有頭來指明消息壓縮類型，故在消費者端消息解壓是透明的無需指定。
compression.codec=none
# 指定序列化處理類
serializer.class=kafka.serializer.DefaultEncoder
# 如果要壓縮消息，這裡指定哪些topic要壓縮消息，預設empty，表示不壓縮。
#compressed.topics=
# 設定發送資料是否需要服務端的回報,有三個值0,1,-1
# 0: producer不會等待broker發送ack
# 1: 當leader接收到消息之後發送ack
# -1: 當所有的follower都同步消息成功後發送ack.
request.required.acks=
# 在向producer發送ack之前,broker允許等待的最大時間 ，如果逾時,broker将會向producer發送一個error ACK.意味着上一次消息因為某種原因未能成功(比如follower未能同步成功)
request.timeout.ms=
# 同步還是異步發送消息，預設“sync”表同步，"async"表異步。異步可以提高發送吞吐量,
也意味着消息将會在本地buffer中,并适時批量發送，但是也可能導緻丢失未發送過去的消息
producer.type=sync
# 在async模式下,當message被緩存的時間超過此值後,将會批量發送給broker,預設為5000ms
# 此值和batch.num.messages協同工作.
queue.buffering.max.ms =
# 在async模式下,producer端允許buffer的最大消息量
# 無論如何,producer都無法盡快的将消息發送給broker,進而導緻消息在producer端大量沉積
# 此時,如果消息的條數達到閥值,将會導緻producer端阻塞或者消息被抛棄，預設為10000
queue.buffering.max.messages=
# 如果是異步，指定每次批量發送資料量，預設為200
batch.num.messages=
# 當消息在producer端沉積的條數達到"queue.buffering.max.meesages"後
# 阻塞一定時間後,隊列仍然沒有enqueue(producer仍然沒有發送出任何消息)
# 此時producer可以繼續阻塞或者将消息抛棄,此timeout值用于控制"阻塞"的時間
# -1: 無阻塞逾時限制,消息不會被抛棄
# 0:立即清空隊列,消息被抛棄
queue.enqueue.timeout.ms=-
# 當producer接收到error ACK,或者沒有接收到ACK時,允許消息重發的次數
# 因為broker并沒有完整的機制來避免消息重複,是以當網絡異常時(比如ACK丢失)
# 有可能導緻broker接收到重複的消息,預設值為3.
message.send.max.retries=
# producer重新整理topic metada的時間間隔,producer需要知道partition leader的位置,以及目前topic的情況
# 是以producer需要一個機制來擷取最新的metadata,當producer遇到特定錯誤時,将會立即重新整理
# (比如topic失效,partition丢失,leader失效等),此外也可以通過此參數來配置額外的重新整理機制，預設值600000
topic.metadata.refresh.interval.ms=
消費者配置檔案解釋
# zookeeper連接配接伺服器位址
zookeeper.connect=zk01:,zk02:,zk03:
# zookeeper的session過期時間，預設5000ms，用于檢測消費者是否挂掉
zookeeper.session.timeout.ms=
#當消費者挂掉，其他消費者要等該指定時間才能檢查到并且觸發重新負載均衡
zookeeper.connection.timeout.ms=
# 指定多久消費者更新offset到zookeeper中。注意offset更新時基于time而不是每次獲得的消息。一旦在更新zookeeper發生異常并重新開機，将可能拿到已拿到過的消息
zookeeper.sync.time.ms=
#指定消費組
group.id=xxx
# 當consumer消費一定量的消息之後,将會自動向zookeeper送出offset資訊
# 注意offset資訊并不是每消費一次消息就向zk送出一次,而是現在本地儲存(記憶體),并定期送出,預設為true
auto.commit.enable=true
# 自動更新時間。預設60 * 1000
auto.commit.interval.ms=
# 目前consumer的辨別,可以設定,也可以有系統生成,主要用來跟蹤消息消費情況,便于觀察
conusmer.id=xxx
# 消費者用戶端編号，用于區分不同用戶端，預設用戶端程式自動産生
client.id=xxxx
# 最大取多少塊緩存到消費者(預設10)
queued.max.message.chunks=
# 當有新的consumer加入到group時,将會reblance,此後将會有partitions的消費端遷移到新 的consumer上,如果一個consumer獲得了某個partition的消費權限,那麼它将會向zk注冊 "Partition Owner registry"節點資訊,但是有可能此時舊的consumer尚沒有釋放此節點, 此值用于控制,注冊節點的重試次數.
rebalance.max.retries=
# 擷取消息的最大尺寸,broker不會像consumer輸出大于此值的消息chunk 每次feth将得到多條消息,此值為總大小,提升此值,将會消耗更多的consumer端記憶體
fetch.min.bytes=
# 當消息的尺寸不足時,server阻塞的時間,如果逾時,消息将立即發送給consumer
fetch.wait.max.ms=
socket.receive.buffer.bytes=
# 如果zookeeper沒有offset值或offset值超出範圍。那麼就給個初始的offset。有smallest、largest、anything可選，分别表示給目前最小的offset、目前最大的offset、抛異常。預設largest
auto.offset.reset=smallest
# 指定序列化處理類
derializer.class=kafka.serializer.DefaultDecoder
![虛拟機---kafka的安裝[圖]](data:image/gif;base64,R0lGODlhAQABAIAAAP///wAAACwAAAAAAQABAAACAkQBADs=)