使用java實現Kafka的消費者
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
|
package
com.lisg.kafkatest;
import
java.util.HashMap;
import
java.util.List;
import
java.util.Map;
import
java.util.Properties;
import
java.util.concurrent.ExecutorService;
import
java.util.concurrent.Executors;
import
java.util.concurrent.TimeUnit;
import
kafka.consumer.Consumer;
import
kafka.consumer.ConsumerConfig;
import
kafka.consumer.ConsumerIterator;
import
kafka.consumer.KafkaStream;
import
kafka.javaapi.consumer.ConsumerConnector;
/**
* java實現Kafka消費者的示例
* @author lisg
*
*/
public
class
KafkaConsumer {
private
static
final
String TOPIC =
"test"
;
private
static
final
int
THREAD_AMOUNT =
1
;
public
static
void
main(String[] args) {
Properties props =
new
Properties();
props.put(
"zookeeper.connect"
,
"vm1:2181"
);
props.put(
"group.id"
,
"group1"
);
props.put(
"zookeeper.session.timeout.ms"
,
"400"
);
props.put(
"zookeeper.sync.time.ms"
,
"200"
);
props.put(
"auto.commit.interval.ms"
,
"1000"
);;
Map<String, Integer> topicCountMap =
new
HashMap<String, Integer>();
//每個topic使用多少個kafkastream讀取, 多個consumer
topicCountMap.put(TOPIC, THREAD_AMOUNT);
//可以讀取多個topic
// topicCountMap.put(TOPIC2, 1);
ConsumerConnector consumer = Consumer.createJavaConsumerConnector(
new
ConsumerConfig(props));
Map<String, List<KafkaStream<
byte
[],
byte
[]>>> msgStreams = consumer.createMessageStreams(topicCountMap );
List<KafkaStream<
byte
[],
byte
[]>> msgStreamList = msgStreams.get(TOPIC);
//使用ExecutorService來調度線程
ExecutorService executor = Executors.newFixedThreadPool(THREAD_AMOUNT);
for
(
int
i =
0
; i < msgStreamList.size(); i++) {
KafkaStream<
byte
[],
byte
[]> kafkaStream = msgStreamList.get(i);
executor.submit(
new
HanldMessageThread(kafkaStream, i));
}
//關閉consumer
try
{
Thread.sleep(
20000
);
}
catch
(InterruptedException e) {
e.printStackTrace();
}
if
(consumer !=
null
) {
consumer.shutdown();
}
if
(executor !=
null
) {
executor.shutdown();
}
try
{
if
(!executor.awaitTermination(
5000
, TimeUnit.MILLISECONDS)) {
System.out.println(
"Timed out waiting for consumer threads to shut down, exiting uncleanly"
);
}
}
catch
(InterruptedException e) {
System.out.println(
"Interrupted during shutdown, exiting uncleanly"
);
}
}
}
/**
* 具體處理message的線程
* @author Administrator
*
*/
class
HanldMessageThread
implements
Runnable {
private
KafkaStream<
byte
[],
byte
[]> kafkaStream =
null
;
private
int
num =
0
;
public
HanldMessageThread(KafkaStream<
byte
[],
byte
[]> kafkaStream,
int
num) {
super
();
this
.kafkaStream = kafkaStream;
this
.num = num;
}
public
void
run() {
ConsumerIterator<
byte
[],
byte
[]> iterator = kafkaStream.iterator();
while
(iterator.hasNext()) {
String message =
new
String(iterator.next().message());
System.out.println(
"Thread no: "
+ num +
", message: "
+ message);
}
}
}
|
|
1
|
props.put(
"auto.commit.interval.ms"
,
"1000"
);
|
表示的是：consumer間隔多長時間在zookeeper上更新一次offset
說明：
為什么使用High Level Consumer？
有些場景下，從Kafka中讀取消息的邏輯不處理消息的offset，僅僅是獲取消息數據。High Level Consumer就提供了這種功能。
首先要知道的是，High Level Consumer在ZooKeeper上保存最新的offset（從指定的分區中讀取）。這個offset基於consumer group名存儲。
Consumer group名在Kafka集群上是全局性的，在啟動新的consumer group的時候要小心集群上沒有關閉的consumer。當一個consumer線程啟動了，Kafka會將它加入到相同的topic下的相同consumer group里，並且觸發重新分配。在重新分配時，Kafka將partition分配給consumer，有可能會移動一個partition給另一個consumer。如果老的、新的處理邏輯同時存在，有可能一些消息傳遞到了老的consumer上。
設計High Level Consumer
使用High LevelConsumer首先要知道的是，它應該是多線程的。消費者線程的數量跟tipic的partition數量有關，它們之間有一些特定的規則：
如果線程數量大於主題的分區數量，一些線程將得不到任何消息
如果分區數大於線程數，一些線程將得到多個分區的消息
如果一個線程處理多個分區的消息，它接收到消息的順序是不能保證的。比如，先從分區10獲取了5條消息，從分區11獲取了6條消息，然后從分區10獲取了5條，緊接着又從分區10獲取了5條，雖然分區11還有消息。
添加更多了同consumer group的consumer將觸發Kafka重新分配，某個分區本來分配給a線程的，從新分配后，有可能分配給了b線程。
關閉消費組和錯誤處理
Kafka不會再每次讀取消息后馬上更新zookeeper上的offset，而是等待一段時間。由於這種延遲，有可能消費者讀取了一條消息，但沒有更新offset。所以，當客戶端關閉或崩潰后，從新啟動時有些消息重復讀取了。另外，broker宕機或其他原因導致更換了partition的leader，也會導致消息重復讀取。
為了避免這種問題，你應該提供一個平滑的關閉方式，而不是使用kill -9
上面的java代碼中提供一種關閉的方式：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
if
(consumer !=
null
) {
consumer.shutdown();
}
if
(executor !=
null
) {
executor.shutdown();
}
try
{
if
(!executor.awaitTermination(
5000
, TimeUnit.MILLISECONDS)) {
System.out.println(
"Timed out waiting for consumer threads to shut down, exiting uncleanly"
);
}
}
catch
(InterruptedException e) {
System.out.println(
"Interrupted during shutdown, exiting uncleanly"
);
}
|
在shutdown之后，等待了5秒鍾，給consumer線程時間來處理完kafka stream里保留的消息。
參考資料：https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example
附件列表