springboot集成kafka
1、引入依赖
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>2.8.6</version>
</dependency>
2、配置
server:
port: 9099
spring:
kafka:
bootstrap-servers: 192.168.157.101:9092
consumer:
group-id: test-consumer-group
max-poll-records: 10
concurrency: 10
#Kafka中没有初始偏移或如果当前偏移在服务器上不再存在时,默认区最新 ,有三个选项 【latest, earliest, none】
auto-offset-reset: earliest
#是否开启自动提交
enable-auto-commit: false
ack-mode: MANUAL_IMMEDIATE
#自动提交的时间间隔
auto-commit-interval: 1000
#key的解码方式
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
#value的解码方式
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
properties:
batch-listener: true
producer:
batch-size: 4096
buffer-memory: 40960
retries: 1
value-serializer: org.apache.kafka.common.serialization.StringSerializer
key-serializer: org.apache.kafka.common.serialization.StringSerializer
listener:
#创建多少个consumer,值必须小于等于Kafk Topic的分区数。
ack-mode: MANUAL_IMMEDIATE
concurrency: 1 #推荐设置为topic的分区数
3、测试
@Autowired
KafkaTemplate<String,String> kafkaTemplate;
@Test
void contextLoads() {
for (int i = 0; i < 100; i++) {
kafkaTemplate.send("hello","hello"+i);
}
}
如日志打印:Connection to node 0 (/127.0.0.1:9092)
#进入kafka安装目录
/usr/local/kafka_2.12-3.5.1/config
#编辑配置文件
vim server.properties
# 允许外部端口连接
listeners=PLAINTEXT://0.0.0.0:9092
# 外部代理地址
advertised.listeners=PLAINTEXT://192.168.157.101:9092
重启kafka服务