Kafka - 新消费者

Myth丶恋晨 2022-05-27 14:10 215阅读 0赞

# Kafka - 新消费者 #

--------------------

# 一、数据来源 #

数据使用上一个[博文][Link 1]所配置的 Flume，将文本数据写入到 Kafka中。不过这次有所改变，数据的监控目录 有所改变，写入的Kafka的主题名也变更为`A25`。

![flume写入数据.png-36kB][flume_.png-36kB]

这里我们可以看到 Flume 对于新传上去的 A91 数据已经完成消费。

# 二、消费者代码 #

# 2.1 创建消费者 #

创建消费者所使用的属性和生产者使用的属性差距不是很大：

1.  bootstrap.servers：指定了 Kafka 集群的连接字符串。
2.  key.deserializer 和 value.deserializer 与生产者的 serializer 定义也很类似，不过它们不是使用指定的类把 Java 对象转成字节数组，而是使用指定的类把字节数组转成 Java 对象。
3.  group.id：非必需，指定了 KafkaConsumer 属于哪一个消费者群组。

创建消费者的代码如下：

Properties props = new Properties();
    props.put("bootstrap.servers", "master:9092");
    props.put("group.id", "TestConsumer");
    props.put("key.deserializer",
            "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer",
            "org.apache.kafka.common.serialization.StringDeserializer");
    
    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

# 2.2 订阅主题 #

consumer.subscribe(Collections.singletonList("A25"));

因为我们使用Flume对数据进行 Sinks 消费的时候，指定的主题为A25，因此我们这里在对数据进行订阅的时候，也是A25。

同时，可以对订阅的主题传递正则表达式进行匹配，一次订阅多个主题。

## 2.3 轮询消费 ##

try {
        while(true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
    
            logger.info("records length = {}", records.count());
    
            for (ConsumerRecord record : records) {
                logger.info("topic = {}, partition = {}, offset = {}, key = {}, value = {}\n",
                        record.topic(), record.partition(), record.offset(),
                        record.key(), record.value());
            }
        }
    } finally {
        consumer.close();
    }

这是一个无限循环。消费者实际上是一个长期运行的应用程序，它通过持续轮询向 Kafka 请求数据。

消费者必须持续对 Kafka 进行轮询，否则会被认为已经死亡，它的分区会被移交给群组里的其他消费者。传给 poll() 方法的参数是一个超时时间，用于控制 poll() 方法的阻塞时间（在消费者的缓冲区里没有可用数据时会发生阻塞）。如果该参数被设为 0，poll() 会立即返回，否则它会在指定的毫秒数内一直等待 broker 返回数据。

poll() 方法返回一个记录列表。每条记录都包含了记录所属主题的信息、记录所在分区的信息、记录在分区里的偏移量，以及记录的键值对。我们一般会遍历这个列表，逐条处理这些记录。

在退出应用程序之前使用 close() 方法关闭消费者。网络连接和 socket 也会随之关闭，并立即触发一次再均衡，而不是等待群组协调器发现它不再发送心跳并认定它已死亡，因为那样需要更长的时间，导致整个群组在一段时间内无法读取消息。

**运行结果如下：**

![kafka消费结果.png-30kB][kafka_.png-30kB]

# 三、其他配置 #

## 3.1 pom文件 ##

<properties>
        <java.version>1.8</java.version>
        <kafka.version>1.1.0</kafka.version>
    </properties>
    
    
    <dependencies>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>${kafka.version}</version>
        </dependency>
    
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.7.21</version>
        </dependency>
    
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.21</version>
        </dependency>
    </dependencies>

## 3.2 log4j.properties ##

log4j.rootLogger=INFO,console
    log4j.additivity.org.apache=true
    
    # 控制台(console)
    log4j.appender.console=org.apache.log4j.ConsoleAppender
    log4j.appender.console.Threshold=DEBUG
    log4j.appender.console.ImmediateFlush=true
    log4j.appender.console.Target=System.err
    log4j.appender.console.layout=org.apache.log4j.PatternLayout
    log4j.appender.console.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} [%p] %c  -  %m%n

[Link 1]: https://blog.csdn.net/u011669700/article/details/79993335
[flume_.png-36kB]: /images/20220527/6e0ddb3183f64dfa9585b5ffa0fbe84e.png
[kafka_.png-30kB]: /images/20220527/0ab0fb5e66734c008f650f838846fd5f.png