1、生產者服務搭建

1.1、引入spring-kafka依賴

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>3.0.5</version><relativePath/> <!-- lookup parent from repository --></parent><!-- Generated by https://start.springboot.io --><!-- 優質的 spring/boot/data/security/cloud 框架中文文檔盡在 => https://springdoc.cn --><groupId>com.atguigu.kafka</groupId><artifactId>kafka-producer</artifactId><version>0.0.1-SNAPSHOT</version><name>kafka-producer</name><description>kafka-producer</description><properties><java.version>17</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

1.2、使用Java代碼創建主題分區副本

package com.atguigu.kafka.config;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.stereotype.Component;
@Component
public class KafkaTopicConfig {@Beanpublic NewTopic myTopic1() {//相同名稱的主題 只會創建一次，后面創建的主題名稱相同配置不同可以做增量更新（分區、副本數）return TopicBuilder.name("my_topic1")//主題名稱.partitions(3)//主題分區.replicas(3)//主題分區副本數.build();//創建}
}

1.3、發送消息

package com.atguigu.kafka.producer;import jakarta.annotation.Resource;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.KafkaTemplate;@SpringBootTest
class KafkaProducerApplicationTests {//裝配kafka模板類： springboot啟動時會自動根據配置文初始化kafka模板類對象注入到容器中@ResourceKafkaTemplate kafkaTemplate;@Testvoid contextLoads() {kafkaTemplate.send("my_topic1", "spring-kafka");}
}

[[{"partition": 0,"offset": 0,"msg": "spring-kafka","timespan": 1717483674110,"date": "2024-06-04 06:47:54"}]
]