java实现Kafka的消费者示例

使用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

来自为知笔记(Wiz)

附件列表

时间： 2024-10-09 23:08:18

java实现Kafka的消费者示例的相关文章

kafka集群搭建和使用Java写kafka生产者消费者

 kafka集群搭建 Java代码   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/kaf

java实现Kafka生产者示例

使用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 package com.lisg.kafkatest; import

Java版Kafka使用及配置解释

Java版Kafka使用及配置解释 一.Java示例 kafka是吞吐量巨大的一个消息系统,它是用scala写的,和普通的消息的生产消费还有所不同,写了个demo程序供大家参考.kafka的安装请参考官方文档. 引入Maven库 首先我们需要新建一个maven项目,然后在pom中引用kafka jar包,引用依赖如下: <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka_2.1

java多线程--“升级版”生产者消费者

ReentrantLock介绍 ReentrantLock是一个可重入的互斥锁,又被称为"独占锁". 顾名思义,ReentrantLock锁在同一个时间点只能被一个线程锁持有:而可重入的意思是,ReentrantLock锁,可以被单个线程多次获取. ReentrantLock分为"公平锁"和"非公平锁".它们的区别体现在获取锁的机制上是否公平."锁"是为了保护竞争资源,防止多个线程同时操作线程而出错,ReentrantLock

Java多线程--生产者与消费者问题

说明 Java中,线程之间的通信主要是由java.lang.Object类提供的wait.notify和notifyAll这3个方法来完成: ①对象的wait方法被调用后,线程进入对象的等待队列中,并释放对象锁,其它线程可以竞争使用此对象锁:sleep方法使得一个线程进入睡眠状态,但是线程所占有的资源并没有释放. ②当对象的notify方法被调用,该方法会从对象的等待队列中随机取出一个线程来唤醒:notifyAll是唤醒等待队列中所有线程,这些线程会与其它正在执行的线程共同竞争对象锁. ③wai

Java里的生产者-消费者模型(Producer and Consumer Pattern in Java)

生产者-消费者模型是多线程问题里面的经典问题,也是面试的常见问题.有如下几个常见的实现方法: 1. wait()/notify() 2. lock & condition 3. BlockingQueue 下面来逐一分析. 1. wait()/notify() 第一种实现,利用根类Object的两个方法wait()/notify(),来停止或者唤醒线程的执行:这也是最原始的实现. 1 public class WaitNotifyBroker<T> implements Broker&

一个非常标准的Java连接Oracle数据库的示例代码

最基本的Oracle数据库连接代码(只针对Oracle11g): 1.右键项目->构建路径->配置构建路径,选择第三项"库",然后点击"添加外部Jar",选择"D:\Oracle\app\oracle\product\11.2.0\server \jdbc\lib\ojdbc6_g.jar"(注:D:\Oracle为数据库的安装路径). 2.以下代码为非常标准的Oracle数据库连接代码示例: /** * 一个非常标准的连接Oracl

JAVA多线程之生产者消费者

生产者消费者并发编程: 假设仓库有10个仓位,分别有10个生产者和10个消费者,生产者不断生产产品,放入仓库的仓位中,而消费者则不断从仓库中获取产品, 如果仓库已满,则生产者要等待,等消费者消费后,空出仓位后,再继续放入产品. 反之如果仓库已空,则消费者要等待,等待生产者生产出产品后,再继续消费产品. 关于生产者.消费者有四种实现方式 1,wait,nofity方式 2,ReentrantLock锁的await()和signal() 3,阻塞队列的方式 4,Semaphore 信号量方式 下面分

Android中Java反射技术的使用示例

import java.lang.reflect.Constructor; import java.lang.reflect.Field; import java.lang.reflect.Method; import android.os.Bundle; import android.app.Activity; /** * Demo描述: * Android中Java反射技术的使用示例 * 在Java中描述字节码文件(xxx.class)的类叫Class * 反射的过程可视为剖析Class的过