kafak集群搭建-基于kRaft方式

1、服务器规划

2、kafka集群部署配置

至少需要三个节点

2.1、解压三个kafka

tar -zxvf kafka_2.13-3.7.0.tgz -C /usr/local/

2.2、配置/config/kraft/server.properties

其中马赛克处需要修改为你的服务器真实IP

配置第一台

配置第二台：

配置第三台：

3、启动kafka集群

1、生成Cluster UUID（集群UUID）

./kafka-storage.sh random-uuid

2、格式化日志目录

./kafka-storage.sh format -t UBLFE9CCRwauv0Mc3tU7qQ -c ../config/kraft/server.properties

3、启动kafka

//后台启动
./kafka-server-start.sh ../config/kraft/server.properties &

4、关闭kafka

./kafka-server-stop.sh ../config/kraft/server.properties

4、SpringBoot集成kafka的kRaft集群

4.1、消费者

package com.power.consumer;import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;@Component
public class EventConsumer {/*** topics 用于指定从哪个主题中消费消息* concurrency 用于指定有多少个消费者* @param record*/@KafkaListener(topics = {"kraftClusterTopic"}, groupId = "kraftclusterGroup")public void onEventA(ConsumerRecord<String, String> record) {System.out.println(Thread.currentThread().getId()+"---> 消费消息 record = " + record);}
}

4.2、生产者

package com.power.producer;import com.power.model.User;
import com.power.util.JSONUtils;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;import javax.annotation.Resource;
import java.util.Date;@Component
public class EventProducer {@Resourceprivate KafkaTemplate<String,Object> kafkaTemplate;public void sendEvent(){for (int i = 0; i < 2; i++) {User user = User.builder().id(i).phone("1567676767"+i).birthday(new Date()).build();String userJson = JSONUtils.toJSON(user);kafkaTemplate.send("kraftClusterTopic","k"+i, userJson);}}}

4.3、配置类

package com.power.config;import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class KafkaConfig {@Beanpublic NewTopic newTopic(){//设置副本个数不能为0，也不能大于节点个数，否则将不能创建Topicreturn new NewTopic("kraftClusterTopic",2, (short)2);}
}

4.4、实体类

package com.power.model;import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;import java.util.Date;@Builder
@AllArgsConstructor
@NoArgsConstructor
@Data
public class User {private Integer id;private String phone;private Date birthday;}

4.5、JSON工具类

package com.power.util;import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;public class JSONUtils {private static final ObjectMapper OBJECTMAPPER = new ObjectMapper();public static String toJSON(Object object){try {return OBJECTMAPPER.writeValueAsString(object);} catch (JsonProcessingException e) {throw new RuntimeException(e);}}public static <T> T toBean(String json,Class<T> clazz){try {return OBJECTMAPPER.readValue(json,clazz);} catch (JsonProcessingException e) {throw new RuntimeException(e);}}
}

4.6、项目配置文件

spring:application:#应用名称name: spring-boot-08-kafka-kRaftCluster#kafka连接地址（ip+port）kafka:bootstrap-servers: <你的服务器IP>:9091,<你的服务器IP>:9092,<你的服务器IP>:9093#配置消费者的反序列化consumer:key-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializer

4.7、测试类

package com.power;import com.power.producer.EventProducer;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;import javax.annotation.Resource;@SpringBootTest
public class SpringBoot08KafkaBaseApplication {@Resourceprivate EventProducer eventProducer;@Testvoid sendInterceptor(){eventProducer.sendEvent();}}