当前位置：首页 > java >正文

kafka快速部署、集成、调优

java 2025/8/6 16:54:30

kafka快速部署、集成、调优

1.部署

1.1 docker部署

参考：https://blog.csdn.net/taotao_guiwang/article/details/135508643
在这里插入图片描述

1.2 kafka部署

资源见，百度网盘：https://pan.baidu.com/s/1qlabJ7m8BDm77GbDuHmbNQ?pwd=41ac
修改配置文件，说明要在那个机器上部署：
执行kafka_zk_install.sh，开始部署：

# 赋权，在kafka_zk_install目录，执行：
chmod -R 777 .
# 执行
./kafka_zk_install.sh

在这里插入图片描述

2. spring boot集成

核心代码，KafkaController.java:

package com.kafka;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;@RestController
public class KafkaController {private final static String TOPIC_NAME = "my-replicated-topic";@Autowiredprivate KafkaTemplate<String, String> kafkaTemplate;@RequestMapping("/send")public void send() {kafkaTemplate.send(TOPIC_NAME, 0, "key", "this is a msg");}}

MyConsumer.java:

package com.kafka;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;@RestController
public class KafkaController {private final static String TOPIC_NAME = "my-replicated-topic";@Autowiredprivate KafkaTemplate<String, String> kafkaTemplate;@RequestMapping("/send")public void send() {kafkaTemplate.send(TOPIC_NAME, 0, "key", "this is a msg");}}

application.yml:

server:port: 8080spring:kafka:bootstrap-servers: 10.86.97.210:9092,10.86.97.210:9093,10.86.97.210:9094producer: # 生产者retries: 3 # 设置大于0的值，则客户端会将发送失败的记录重新发送batch-size: 16384buffer-memory: 33554432acks: 1# 指定消息key和消息体的编解码方式key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializerconsumer:group-id: default-groupenable-auto-commit: falseauto-offset-reset: earliestkey-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializerlistener:# 当每一条记录被消费者监听器（ListenerConsumer）处理之后提交# RECORD# 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后提交# BATCH# 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交# TIME# 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交# COUNT# TIME |　COUNT　有一个条件满足时提交# COUNT_TIME# 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgment.acknowledge()后提交# MANUAL# 手动调用Acknowledgment.acknowledge()后立即提交# MANUAL_IMMEDIATEack-mode: MANUAL_IMMEDIATE