kafka创建topic_SpringBoot集成kafka全面实战

本文是SpringBoot+Kafka的实战讲解，如果对kafka的架构原理还不了解的读者，建议先看一下《大白话kafka架构原理》、《秒懂kafka HA(高可用)》两篇文章。

一、生产者实践

• 普通生产者

• 带回调的生产者

• 自定义分区器

• kafka事务提交

二、消费者实践

• 简单消费

• 指定topic、partition、offset消费

• 批量消费

• 监听异常处理器

• 消息过滤器

• 消息转发

• 定时启动/停止监听器

0 1前戏 1、开放 advertised.listeners配置

在项目中连接kafka，因为是外网，首先要开放kafka配置文件中的如下配置(其中IP为公网IP)，

advertised.listeners=PLAINTEXT://112.126.74.249:9092

2、创建topic 在开始前我们先创建两个topic：topic1、topic2，其分区和副本数都设置为2，用来测试，

[root@iZ2zegzlkedbo3e64vkbefZ ~]#  cd /usr/local/kafka-cluster/kafka1/bin/[root@iZ2zegzlkedbo3e64vkbefZ bin]# ./kafka-topics.sh --create --zookeeper 172.17.80.219:2181 --replication-factor 2 --partitions 2 --topic topic1Created topic topic1.[root@iZ2zegzlkedbo3e64vkbefZ bin]# ./kafka-topics.sh --create --zookeeper 172.17.80.219:2181 --replication-factor 2 --partitions 2 --topic topic2Created topic topic2.

当然我们也可以不手动创建topic，在执行代码kafkaTemplate.send("topic1", normalMessage)发送消息时，kafka会帮我们自动完成topic的创建工作，但这种情况下创建的topic默认只有一个分区，分区也没有副本。所以，我们可以在项目中新建一个配置类专门用来初始化topic，如下，

@Configurationpublic class KafkaInitialConfiguration {
        // 创建一个名为testtopic的Topic并设置分区数为8，分区副本数为2    @Bean    public NewTopic initialTopic() {
            return new NewTopic("testtopic",8, (short) 2 );    }     // 如果要修改分区数，只需修改配置值重启项目即可    // 修改分区数并不会导致数据的丢失，但是分区数只能增大不能减小    @Bean    public NewTopic updateTopic() {
            return new NewTopic("testtopic",10, (short) 2 );    }}

3、新建SpringBoot项目

① 引入pom依赖

<dependency>    <groupId>org.springframework.kafkagroupId>    <artifactId>spring-kafkaartifactId>dependency>

② application.propertise配置(本文用到的配置项这里全列了出来)

###########【Kafka集群】###########spring.kafka.bootstrap-servers=112.126.74.249:9092,112.126.74.249:9093###########【初始化生产者配置】############ 重试次数spring.kafka.producer.retries=0# 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)spring.kafka.producer.acks=1# 批量大小spring.kafka.producer.batch-size=16384# 提交延时spring.kafka.producer.properties.linger.ms=0# 当生产端积累的消息达到batch-size或接收到消息linger.ms后,生产者就会将消息提交给kafka# linger.ms为0表示每接收到一条消息就提交给kafka,这时候batch-size其实就没用了# 生产端缓冲区大小spring.kafka.producer.buffer-memory = 33554432# Kafka提供的序列化和反序列化类spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializerspring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer# 自定义分区器# spring.kafka.producer.properties.partitioner.class=com.felix.kafka.producer.CustomizePartitioner###########【初始化消费者配置】############ 默认的消费组IDspring.kafka.consumer.properties.group.id=defaultConsumerGroup# 是否自动提交offsetspring.kafka.consumer.enable-auto-commit=true# 提交offset延时(接收到消息后多久提交offset)spring.kafka.consumer.auto.commit.interval.ms=1000# 当ka