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 {

    @Autowired
    KafkaTemplate kafkaTemplate;

    @Test
    void 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 {

    @Autowired
    KafkaTemplate kafkaTemplate;

    @Test
    void 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

• 测试成功