6.4 消息发送服务

---

• 访问 kafka-ui （注意这里需要换成你自己的服务器或者虚拟机的 IP 地址，虚拟机可以用局域网 192.168.xxx.xxx 的地址）

http://192.168.1.4:8080/ui

• 创建主题

• 创建成功

• application.properties 配置文件（注意需要改成你的 kafka 地址，也就是浏览器访问的主机 IP ）

spring.kafka.bootstrap-servers=192.168.1.4:9092

• 测试代码

package com.zhong.message;import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.util.StopWatch;import java.util.concurrent.CompletableFuture;@SpringBootTest
class Boot309MessageApplicationTests {@AutowiredKafkaTemplate kafkaTemplate;@Testvoid contextLoads() {// 秒表功能StopWatch stopWatch = new StopWatch();// 异步处理等待处理完CompletableFuture[] futures = new CompletableFuture[10000];// 开始计时stopWatch.start();for (int i = 0; i < 10000; i++) {// 异步的CompletableFuture future = kafkaTemplate.send("newshh", "name", "小钟");futures[i] = future;}CompletableFuture.allOf(futures).join();// 停止计时stopWatch.stop();// 统计计时从开始到结束用了多少毫秒long millis = stopWatch.getTotalTimeMillis();System.out.println("10000个消息发送完成！ms时间：" + millis);}}

• 运行测试

• 发送成功

• 新建 Person 类

package com.zhong.message.entity;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;import java.io.Serializable;
import java.util.Date;/*** @ClassName : Person* @Description :* @Author : zhx* @Date: 2024-06-14 16:11*/
@Data
@AllArgsConstructor
@NoArgsConstructor
public class Person implements Serializable {private Long id;private String name;private Integer age;private String email;
}

• 新建测试代码

package com.zhong.message;import com.zhong.message.entity.Person;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.util.StopWatch;import java.util.concurrent.CompletableFuture;@SpringBootTest
class Boot309MessageApplicationTests {@AutowiredKafkaTemplate kafkaTemplate;@Testvoid send() {CompletableFuture future = kafkaTemplate.send("newsPerson", "person", new Person(1L, "小王", 21, "testemal"));future.join();System.out.println("消息发送成功！");}}

• 如果发送出现问题请配置值的序列化规则

# 值的序列化规则
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer

• 测试成功