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
两种方式
- 手动创建
- 我们可以在项目中新建一个配置类专门用来初始化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测试生产消费