Spring-Springkafka基本用法

Kafka - MQ

字数统计: 3.4k阅读时长: 15 min

 2019/04/09 

简介

跟 Spring Data Redis、Spring Data MongoDB、Spring Data JPA 等项目类似，Spring Kafka 提供了在 Spring 应用中通过简单配置从而访问 Kafka 集群的途径。

本文主要介绍在 Spring 应用中消息生产者如何向 Kafka 集群发送消息、消息消费者如何消费消息、如何批量消费消息以及多消费者组同时消费消息等等。

使用 Spring Kafka 的最新特性，以下测试代码采用了 Spring Boot 2.0.0 构建

Spring Kafka 的基本用法

在 pom.xml 中添加依赖：

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

基本配置

springBoot properties

#kafka，更多配置：org.springframework.boot.autoconfigure.kafka.KafkaProperties
#指定kafka 代理地址，可以多个
spring.kafka.bootstrap-servers=127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094
#指定默认topic id
spring.kafka.template.default-topic=topic-test
#指定listener 容器中的线程数，用于提高并发量
spring.kafka.listener.concurrency=3

#=============== provider  =======================
#生产者重试次数
spring.kafka.producer.retries=0
#每次批量发送消息的数量 16K
spring.kafka.producer.batch-size=16384
# 32M
spring.kafka.producer.buffer-memory=33554432
#指定消息key和消息体的编解码方式
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
 
 #=============== consumer  =======================
#指定默认消费者group id
spring.kafka.consumer.group-id=myGroup1
#若设置为earliest，那么会从头开始读partition
spring.kafka.consumer.auto-offset-reset=latest
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

普通 Maven 构建项目，或者想要自定义更多配置，可以采用 JavaConfig 配置

    /**
     *  生产者配置信息
     */
    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = Maps.newHashMap();
        props.put(ProducerConfig.ACKS_CONFIG, "0");
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.RETRIES_CONFIG, retries);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
        props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return props;
    }

    /**
     *  生产者工厂
     */
    @Bean
    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    /**
     *  生产者模板
     */
    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    /**
     *  消费者配置信息
     */
    @Bean
     public Map<String, Object> consumerConfigs() {
        Map<String, Object> propsMap = new HashMap<>();
        propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, propsConfig.getBroker());
        propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, propsConfig.getEnableAutoCommit());
        propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");
        propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
        propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, propsConfig.getGroupId());
        propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, propsConfig.getAutoOffsetReset());
        propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);
        return propsMap;
    }

    /**
     *  消费者批量工厂
     */
    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(4);
        //设置为批量消费，每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_CONFIG
        factory.setBatchListener(true);
        factory.getContainerProperties().setPollTimeout(3000);
        return factory;
    }
}

详细含义参考官方文档： https://kafka.apache.org/documentation/#producerapi

第一个消息生产 / 消费的实例

SimpleProducer ：

@Component
public class SimpleProducer {

    private static final Logger logger = LoggerFactory.getLogger(SimpleProducer.class);

    @Resource
    private KafkaTemplate<Object, Object> kafkaTemplate;

    /**
     * 向topic中发送消息
     */
    public void send (String topic, String msg) {
        try {
            kafkaTemplate.send(topic, msg);
            logger.info("推送数据成功！");
        } catch (Exception e) {
            logger.error(MessageFormat.format("推送数据出错，topic:{},data:{}",topic,msg));
        }
    }

    /**
     * 向topic中发送消息
     */
    public void send (String topic, List<String> msgs) {
        msgs.forEach(msg -> kafkaTemplate.send(topic, msg));
    }
}

SimpleConsumer ：

@Component
public class SimpleConsumer {

    private static final Logger logger = LoggerFactory.getLogger(SimpleProducer.class);

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    @KafkaListener(id = "test", topics = {"topicName"})
    public void listen(String data) {

        System.out.println("SimpleConsumer收到消息：" + data);
        logger.info(MessageFormat.format("SimpleConsumer收到消息：{}", data));
    }
}

批量消费消息

如果生产者写入消息的速度比消费者读取的速度快的情况下，随着时间增长，消息堆积会越来越严重。
对于这种场景，我们需要增加多个消费者来进行水平扩展。
Kafka 消费者是消费组的一部分，当多个消费者形成一个消费组来消费主题时，每个消费者会收到不同分区的消息。

假设有一个 T1 主题，该主题有 4 个分区；同时我们有一个消费组 G1，这个消费组只有一个消费者 C1。那么消费者 C1 将会收到这 4 个分区的消息，如下所示：

如果我们增加新的消费者 C2 到消费组 G1，那么每个消费者将会分别收到两个分区的消息，如下所示：

如果增加到 4 个消费者，那么每个消费者将会分别收到一个分区的消息，如下所示：

但如果我们继续增加消费者到这个消费组，剩余的消费者将会空闲，不会收到任何消息：

总而言之，我们可以通过增加消费组的消费者来进行水平扩展提升消费能力。这也是为什么建议创建主题时使用比较多的分区数，这样可以在消费负载高的情况下增加消费者来提升性能。另外，消费者的数量不应该比分区数多，因为多出来的消费者是空闲的，没有任何帮助

Kafka 一个很重要的特性就是，只需写入一次消息，可以支持任意多的应用读取这个消息。
换句话说，每个应用都可以读到全量的消息。
为了使得每个应用都能读到全量消息，应用需要有不同的消费组。
对于上面的例子，假如我们新增了一个新的消费组 G2，而这个消费组有两个消费者，那么会是这样的：

在这个场景中，消费组 G1 和消费组 G2 都能收到 T1 主题的全量消息，在逻辑意义上来说它们属于不同的应用。

最后，总结起来就是：
如果应用需要读取全量消息，那么请为该应用设置一个消费组；
如果该应用消费能力不足，那么可以考虑在这个消费组里增加消费者。

监听指定Topic的partition0，代码如下：

@KafkaListener(id = "id0", topicPartitions = {@TopicPartition(topic = TOPIC, partitions = {"0"})})
   public void listenPartition0(List<ConsumerRecord<?, ?>> records) {
       log.info("Id0 Listener, Thread ID: " + Thread.currentThread().getId());
       log.info("Id0 records size " + records.size());

       for (ConsumerRecord<?, ?> record : records) {
           Optional<?> kafkaMessage = Optional.ofNullable(record.value());
           log.info("Received: " + record);
           if (kafkaMessage.isPresent()) {
               Object message = record.value();
               String topic = record.topic();
               log.info("p0 Received message={},topic={}", message,topic);
           }
       }
   }

Demo解析

地址： http://git.blz.netease.com/june.wang/springKafka-demo.git

如何创建消费者

读取 Kafka 消息只需要创建一个 kafkaConsumer，创建过程与 KafkaProducer 非常相像。
我们需要使用四个基本属性，bootstrap.servers、key.deserializer、value.deserializer 和 group.id。
其中，bootstrap.servers 与创建 KafkaProducer 的含义一样；
key.deserializer 和 value.deserializer 是用来做反序列化的，也就是将字节数组转换成对象；
group.id 不是严格必须的，但通常都会指定，这个参数是消费者的消费组。

Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092,broker2:9092");
props.put("group.id", "CountryCounter");
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<String,String>(props);

上面的例子中只设置了几个最基本的消费者参数，bootstrap.servers，group.id，key.deserializer 和 value.deserializer，其他的参数可以看Kafka文档。
虽然我们很多情况下只是使用默认设置就行，但了解一些比较重要的参数还是很有帮助的。
一些比较重要的参数：

订阅主题

1	`consumer.subscribe(Collections.singletonList("topicName"));`

循环拉取

消费数据的 API 和处理方式很简单，我们只需要循环不断拉取消息即可。Kafka 对外暴露了一个非常简洁的 poll 方法，其内部实现了协作、分区重平衡、心跳、数据拉取等功能，但使用时这些细节都被隐藏了，我们也不需要关注这些。

需要提醒的是，消费者对象不是线程安全的，也就是不能够多个线程同时使用一个消费者对象；而且也不能够一个线程有多个消费者对象。简而言之，一个线程一个消费者，如果需要多个消费者那么请使用多线程来进行一一对应。

提交（commit）与位移（offset）

当我们调用 poll () 时，该方法会返回我们没有消费的消息。
当消息从 broker 返回消费者时，broker 并不跟踪这些消息是否被消费者接收到；
Kafka 让消费者自身来管理消费的位移（offset），并向消费者提供更新位移的接口，这种更新位移方式称为提交（commit）。

在正常情况下，消费者会发送分区的提交信息到 Kafka，Kafka 进行记录。
当消费者宕机或者新消费者加入时，Kafka 会进行重平衡，这会导致消费者负责之前并不属于它的分区。
重平衡完成后，消费者会重新获取分区的位移，下面来看下两种有意思的情况。

假如一个消费者在重平衡前后都负责某个分区，如果提交位移比之前实际处理的消息位移要小，那么会导致消息重复消费。

假如在重平衡前某个消费者拉取分区消息，在进行消息处理前提交了位移，但还没完成处理宕机了，然后 Kafka 进行重平衡，新的消费者负责此分区并读取提交位移，此时会 “丢失”消息

因此，提交位移的方式会对应用有比较大的影响

自动确认offset ：

public class AutoCommitConsumerDemo {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.106.203:9092");
        props.put("group.id", "autoCommitConsumers_group");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        @SuppressWarnings("resource")
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("test"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
        }
    }
}

自动提交 offset 的方式非常简单，但多数情况下，我们不会使用自动提交的方式。
因为不论从 Kafka 集群中拉取的数据是否被处理成功，offset 都会被更新，也就是如果处理过程中出现错误可能会出现数据丢失的情况。所以多数情况下我们会选择手动提交方式

手动提交offset：

public class ManualCommitConsumerDemo {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.106.203:9092");
        props.put("group.id", "manualCommitConsumers_group");
        props.put("enable.auto.commit", "false");
        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);
        consumer.subscribe(Collections.singletonList("topic02"));
        final int minBatchSize = 200;
        List<ConsumerRecord<String, String>> buffer = new ArrayList<>();
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> record : records) {
                buffer.add(record);
            }
            if (buffer.size() >= minBatchSize) {
                //do something
                insertIntoDb(buffer);
                //异步提交，防止阻塞进程。但是异步提交也有个缺点，那就是如果服务器返回提交失败，异步提交不会进行重试（避免多个异步提交，导致位移覆盖）。
                //consumer.commitSync();

                //一般情况下对于异步提交，我们可能会通过回调的方式记录提交结果
                consumer.commitAsync((offsets, e) -> {
                    if (e != null) {
                        log.error("Commit failed for offsets {}", offsets, e);
                    }
                });

                buffer.clear();
            }
        }
    }

    private static void insertIntoDb(List<ConsumerRecord<String, String>> buffer) {
        System.out.println(buffer);
        // Insert into db
    }

}

KafkaConsumer从指定位移（offset）开始消费。：

@Override
public void seek(TopicPartition partition, long offset) {
    if (offset < 0)
        throw new IllegalArgumentException("seek offset must not be a negative number");

    acquireAndEnsureOpen();
    try {
        log.debug("Seeking to offset {} for partition {}", offset, partition);
        this.subscriptions.seek(partition, offset);
    } finally {
        release();
    }
}

另外注意的是，seek () 只是指定了 poll () 拉取的开始位移，这并不影响在 Kafka 中保存的提交位移（当然我们可以在 seek 和 poll 之后提交位移覆盖）。

优雅退出

在一般情况下，我们会在一个主线程中循环 poll 消息并进行处理。
当需要退出 poll 循环时，我们可以使用另一个线程调用 consumer.wakeup ()，调用此方法会使得 poll () 抛出 WakeupException。如果调用 wakup 时，主线程正在处理消息，那么在下一次主线程调用 poll 时会抛出异常WakeUpException。
主线程在抛出 WakeUpException 后，需要调用 consumer.close ()，此方法会提交位移，同时发送一个退出消费组的消息到 Kafka 的组协调者。组协调者收到消息后会立即进行重平衡（而无需等待此消费者会话过期）。

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(1000);
        System.out.println(System.currentTimeMillis() + "--  waiting for data...");
        for (ConsumerRecord<String, String> record : records) {
            System.out.printf("offset = %d, key = %s, value = %s\n",record.offset(), record.key(), record.value());
        }
        for (TopicPartition tp: consumer.assignment())
            System.out.println("Committing offset at position:" + consumer.position(tp));
        consumer.commitSync();
    }
} catch (WakeupException e) {
    log.error("", e);
} finally {
    consumer.close();
    System.out.println("Consumer Closed");
}