Kafka整合springboot

1、在项目中连接kafka

因为是外网，首先要开放kafka配置文件中的如下配置（其中IP为公网IP），（本人这里用的本地）在kafka文件夹下面的config下的server.properties文件。

2.springboot中导入kafka依赖

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

3.初始化topic

两种方式

1. 手动创建
2. 我们可以在项目中新建一个配置类专门用来初始化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 );
    }
}

4.配置文件

• 在application.yaml文件中如下

server:
  port: 8081
spring:
  application:
    name: kafka-ts
  # 指定kafka 代理地址，可以多个
  kafka:
    bootstrap-servers: 127.0.0.1:9092
    producer: #生产者
      retries: 3 #重试次数
      batch-size: 16384 #每次批量发送消息的数量16384
      buffer-memory: 3354432
      key-serializer: org.apache.kafka.common.serialization.StringSerializer #指定消息key和消息体的编解码方式
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer: #消费者
      group-id: ts-log-group
      auto-offset-reset: earliest
      enable-auto-commit: true
      auto-commit-interval: 100
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

• 在application.properties文件中如下

spring.application.name=kafka-ts
server.port=8081
#============== kafka ===================
# 指定kafka 代理地址，可以多个
spring.kafka.bootstrap-servers=127.0.0.1:9092
#=============== provider  =======================
spring.kafka.producer.retries=3
# 每次批量发送消息的数量
spring.kafka.producer.batch-size=16384
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=ts-log-group
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.enable-auto-commit=true
spring.kafka.consumer.auto-commit-interval=100

# 指定消息key和消息体的编解码方式
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

5.生产者和消费者

简单生产者

@RestController
public class KafkaProducer {
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    // 发送消息
    @GetMapping("/kafka/normal/{message}")
    public void sendMessage1(@PathVariable("message") String normalMessage) {
        kafkaTemplate.send("topic1", normalMessage);
    }
    @GetMapping("/test")
    public void send(){
        System.out.println("11111");
    }

}

简单消费者

@Component
public class KafkaConsumer {
    // 消费监听
    @KafkaListener(topics = {"topic1"})
    public void onMessage1(ConsumerRecord<?, ?> record){
        // 消费的哪个topic、partition的消息,打印出消息内容
        System.out.println("简单消费："+record.topic()+"-"+record.partition()+"-"+record.value());
    }

}

6.启动zookeeper和kafka

打开zkServer.cmd和zkCli.cmd，保证zookeeper运行正常,窗口不关闭。

打开kafka目录下cmd，

输入.\bin\windows\kafka-server-start.bat .\config\server.properties

7.启动idea，postman或者jmeter测试生产消费