SpringBoot整合Kafka 上

本文主要记录SpringBoot与Kafka的整合,简单示例
前提Linux上已经搭建好kafka
首先添加依赖:

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

编写yml配置文件:

代码如下:

spring:
  datasource:
    username: 
    password: 
    url:
    driver-class-name: com.mysql.cj.jdbc.Driver
    type: com.alibaba.druid.pool.DruidDataSource
  mvc:
    pathmatch:
      matching-strategy: ant_path_matcher
  kafka:
    ###########【Kafka集群】###########
    bootstrap-servers: 192.168.188.128:9092
    producer:
      retries: 0 # 重试次数
      acks: 1 # 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)
      batch-size: 16384 # 批量大小
      buffer-memory: 33554432 # 生产端缓冲区大小
      # Kafka提供的序列化和反序列化类
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

    consumer:
      group-id: javagroup
      enable-auto-commit: true
      auto-commit-interval: 1000
      # earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
      # latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
      # none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
      auto-offset-reset: latest
      key-deserializer:  org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer:  org.apache.kafka.common.serialization.StringDeserializer

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

advertised.listeners=PLAINTEXT://IP:9092
我的是

代码如下:

listeners=PLAINTEXT://192.168.188.128:9092
advertised.listeners=PLAINTEXT://192.168.188.128:9092

如果把两个IP地址换为

注意:
此时虽然可以用虚拟机创建Topic并收发消息但是IDEA本地会报错如下:

经过多次调试发现只有server.properties

listeners=PLAINTEXT://192.168.188.128:9092
advertised.listeners=PLAINTEXT://192.168.188.128:9092

且本地为:

bootstrap-servers: 192.168.188.128:9092

Linux虚拟机以及IDEA才能正常工作,三者任一都不可为

:localhost:9092

然后新建一个配置类专门用来初始化topic，如下，

代码如下:

@Configuration
public 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);
    }
}

创建Producer:

@RestController
public class KafkaProducer {
    @Autowired(required = false)
    private KafkaTemplate<String,Object> kafkaTemplate;

    @GetMapping("Producer")
    public void sendMessage(String normalMessage){
        kafkaTemplate.send("topic1",normalMessage);
    }
}

创建Consumer:

@Component
public class KafkaConsumer {
    //消费监听
    @KafkaListener(topics = {"topic1"})
    public void onMessage1(ConsumerRecord<?,?>record){
        System.out.println("简单消费: "+record.topic()+"_"+record.partition()+"_"+record.value());
    }
}

使用ApiPost测试

可以看到成功监听:

注意:如果出现:
InvalidReplicationFactorException: Replication factor: 2 larger than available
报错意思是说你当前集群是单节点，无法创建两个副本.