kafka java程式碼編寫
上篇,我介紹瞭如何搭建kafka,但在如何使用kafka上,還沒做過去但介紹,大多人都是寫改main方法去傳送和接受,但我們實際業務肯定不是這樣但。我們公司是以一種方式,類似一直啟動main然後接受訊息的。這裡我也還沒去具體瞭解,之後我會再看看,瞭解之後可能會再寫一篇文章出來。
kafka啟動之後,需要建立topic和partitions,java程式碼只能幫你傳遞訊息和接受訊息,這建立topic和partitions的工作還是得有你自己親自完成
topic相當於一個主題,broker相當於伺服器。比如建立一個日誌的topic,傳送日誌訊息 ,建立一個訂單的topic,傳送訂單資訊。
當然topic裡還有分割槽(partition)的概念,由於一個topic可能太大,利用分割槽更好的接受和傳送訊息,也可以用不同的consumer去消費一個topic下不同partition裡的資料,分割槽規則可以自定義。
每個partition對應一個索引檔案,每個索引檔案對應一個日誌檔案,日誌裡有seq,稀疏索引的主鍵。。(萬一kafka掛了,可以讀這個檔案,用二分法找到自己之前所讀取的那條kafka資料)
像之前一樣,進到kafka目錄,cd /home/kafka
執行以下命令
bin/kafka-topics.sh --create --zookeeper h1:2181,h2:2181,h3:2181 --partition 3 --replication-factor 3 --topic 'test3'
然後你跑到 cd /home/kafka/kafka-logs就會看到有3個檔案生成了
replication-factor 是在幾個broke上生產副本,partition是代表一個topic的分割槽檔案
我這裡建立了一個wmtest2的topic,副本數是1 --replication-factor1,就等於沒有副本,就一個檔案,partition是3,整句話執行完之後是zk會再每個broker上建立一個wmtest2的topic,但每個機器存的分割槽不一樣(如我現在三臺虛擬機器h1,h2,h3 建立之後,wmtest2-0 0號partition放在h1裡,wmtest2-1 1號partition放在h2裡,wmtest2-2 2號partition放在h3裡)
我這裡又多建了幾個topic,為了讓讀者能更深刻的理解
3個分割槽,2個備份
1個分割槽,2個備份
2個分割槽,2個備份
以下是h1伺服器的topic情況,我這裡是建立了3個log檔案,所以topic會分配到上檔案裡的
以下是h2伺服器的topic情況
以下是h3伺服器topic情況
也可以根據命令去查詢
cd到kafka目錄,然後輸入以下命令
bin/kafka-topics.sh --describe --zookeeper h1:2181 --topic 'wmtest2'
Leader: 如果有多個brokerBroker儲存同一個topic,那麼同時只能有一個Broker負責該topic的讀寫,其它的Broker作為實時備份。負責讀寫的Broker稱為Leader.
Replicas : 表示該topic的幾號分割槽在哪幾個broker中儲存
Isr : 表示當前有效的broker, Isr是Replicas的子集
現在殺掉一個broker,操作如下
切到第三臺機器,為這裡是h3,jps一下
然後殺死程序,再看下kafka topic 'wmtest5'的情況
Isr:現在有效的broker就是2了,3已經沒了
本來partition1的leader是3,也變成2了
再啟動你就發現,Isr有效的broker中3回來了,但是leader還是2
http://blog.csdn.net/wangjia184/article/details/37921183這篇文章也詳細介紹了,我所說的東西,還不錯
-----------------------現在開始講解專案工程
首先建一個maven工程
pom檔案如下
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>0.9.0.0</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.10</artifactId>
<version>0.9.0.0</version>
</dependency>
就引入這些就可以了
我這裡一開始也按照網上流程,那樣搭建,跑是能跑起來,但有些我需要模擬但就跑不起來了
一共要建5個類
package com.wm.utilm2;
import java.util.ArrayList;
import java.util.List;
/**
* Created by wangmiao on 2017/6/9.
*/
public class ConsumerGroup {
private List<ConsumerRunnable> consumers;
public ConsumerGroup(int consumerNum, String groupId, String topic, String brokerList) {
consumers = new ArrayList<ConsumerRunnable>(consumerNum);
for (int i = 0; i < consumerNum; ++i) {
ConsumerRunnable consumerThread = new ConsumerRunnable(brokerList, groupId, topic);
consumers.add(consumerThread);
}
}
public void execute() {
for (ConsumerRunnable task : consumers) {
new Thread(task).start();
}
}
}
package com.wm.utilm2;
/**
* Created by wangmiao on 2017/6/9.
*/
public class ConsumerMain {
public static void main(String[] args) {
String brokerList = "h1:9092,h2:9092,h3:9092";
String groupId = "testGroup1";
String topic = "test3";
int consumerNum = 1;
ConsumerGroup consumerGroup = new ConsumerGroup(consumerNum, groupId, topic, brokerList);
consumerGroup.execute();
}
}
package com.wm.utilm2;
/**
* Created by wangmiao on 2017/6/9.
*/
public class ConsumerMainTwo {
public static void main(String[] args) {
String brokerList = "h1:9092,h2:9092,h3:9092";
String groupId = "testGroup2";
String topic = "test3";
int consumerNum = 1;
ConsumerGroup consumerGroup = new ConsumerGroup(consumerNum, groupId, topic, brokerList);
consumerGroup.execute();
}
}
package com.wm.utilm2;
import kafka.consumer.*;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.util.CollectionUtils;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
/**
* Created by wangmiao on 2017/6/8.
*/
public class ConsumerRunnable implements Runnable {
// 每個執行緒維護私有的KafkaConsumer例項
private final KafkaConsumer<String, String> consumer;
public ConsumerRunnable(String brokerList, String groupId, String topic) {
Properties props = new Properties();
props.put("bootstrap.servers", brokerList);
props.put("group.id", groupId);
props.put("enable.auto.commit", "true"); //本例使用自動提交位移
props.put("auto.commit.interval.ms", "1000");
props.put("session.timeout.ms", "30000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
this.consumer = new KafkaConsumer<String, String>(props);
consumer.subscribe(Arrays.asList(topic)); // 本例使用分割槽副本自動分配策略
}
public void run() {
while (true) {
ConsumerRecords<String, String> records = consumer.poll(200); // 本例使用200ms作為獲取超時時間
for (ConsumerRecord<String, String> record : records) {
// 這裡面寫處理訊息的邏輯,本例中只是簡單地列印訊息
System.out.println(Thread.currentThread().getName() + " consumed " + record.partition() +
" ------message is :" + new String(record.value()));
}
}
}
}
package com.wm.util;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import kafka.serializer.StringEncoder;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
import java.util.Random;
/**
* Created by wangmiao on 2017/6/8.
*/
public class KafkaProducerTest extends Thread {
private String topic;
public KafkaProducerTest(String topic){
super();
this.topic = topic;
}
@Override
public void run() {
KafkaProducer<String,String> producer =createProducer();
int i=0;
while(true) {
String msg = "test wm shuai"+i++;
ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(topic, msg);
producer.send(producerRecord);
System.out.println("send " + msg);
try {
Thread.sleep(5000);
}catch (Exception e){
e.printStackTrace();
}
}
}
private KafkaProducer createProducer(){
Properties props = new Properties();
// props.setProperty("zookeeper.connect","h1:2181,h2:2181,h3:2181");
// props.setProperty("serializer.class", StringEncoder.class.getName());
// props.setProperty("metadata.broker.list","h1:9092,h2:9092,h3:9092");
// props.setProperty("partitioner.class","com.wm.util.PersonalPartition");
props.put("bootstrap.servers", "h1:9092,h2:9092,h3:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
return new KafkaProducer(props);
}
public static void main(String arg[]){
new KafkaProducerTest("test1").start();
}
}
ConsumerMainTwo 和 ConsumerMain 是模擬兩個使用者組,但每個使用者組只有一個使用者去消費資訊(同時啟動兩個main方法看下消費訊息的情況)
可以看到不同的使用者組,可以消費同一個topic資訊,
可以改變 consumerNum 來模擬一個使用者組裡的使用者數,但這個使用者數不能超過partition的數量,不然又得使用者是收不到訊息的,因為會被組裡的其他使用者給消費掉。
當之只有一個使用者的時候,消費掉水partition1,2,3但是有兩個使用者組的時候訊息會被分配一個使用者消費兩條,一個使用者消費1條。
如何模擬呢:ConsumerMain的consumerNum設定為1,啟動兩個consumerNum,看到消費情況,然後再kill掉一個,再看看消費情況。
使用者組的幾種情況就完美模擬了