1 pom

<?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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.gzz</groupId>
    <artifactId>03-kafka</artifactId>
    <version>1.0.0</version>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>3.0.3</version>
    </parent>
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <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-json</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

2 application.yml

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      properties.linger.ms: 0
    consumer:
      value-deserializer: org.apache.kafka.common.serialization.ByteArrayDeserializer
      enable-auto-commit: true
      auto.commit.interval.ms: 1

3 Config

package com.gzz.config;

import java.util.Map;

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.support.converter.JsonMessageConverter;
import org.springframework.kafka.support.converter.RecordMessageConverter;
import org.springframework.kafka.support.mapping.DefaultJackson2JavaTypeMapper;
import org.springframework.kafka.support.mapping.Jackson2JavaTypeMapper.TypePrecedence;

import com.gzz.service.User;

/**
 * @author https://www.jianshu.com/u/3bd57d5f1074
 * @date 2023-02-27 14:50:00
 */
@Configuration
public class Config {

    public static final String TOPIC_TEST = "sinosun";
    public static final String TOPIC_GROUP1 = "topic.group1";

    @Bean
    public NewTopic batchTopic() {
        return new NewTopic(TOPIC_TEST, 2, (short) 1);
    }

    @Bean
    public RecordMessageConverter converter() {
        JsonMessageConverter converter = new JsonMessageConverter();
        DefaultJackson2JavaTypeMapper typeMapper = new DefaultJackson2JavaTypeMapper();
        typeMapper.setTypePrecedence(TypePrecedence.TYPE_ID);
        typeMapper.addTrustedPackages("com.gzz.*");//配置扫描路径,多个需要序列化的实体类
        typeMapper.setIdClassMapping(Map.of("abcd_user", User.class));//名子随变取:指定需要序列实体类
        converter.setTypeMapper(typeMapper);
        return converter;
    }
}

4 KafkaService

package com.gzz.service;

import java.time.LocalDateTime;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import com.gzz.config.Config;
import lombok.extern.slf4j.Slf4j;

/**
 * @summary 在一个程序中使用生产者和消费者
 * @author https://www.jianshu.com/u/3bd57d5f1074
 * @date 2023-02-27 14:50:00
 */
@Slf4j
@Component
public class KafkaService {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @KafkaListener(topics = Config.TOPIC_TEST, groupId = Config.TOPIC_GROUP1)
    public void consumer(User user) {
        log.info("user={}", user);
    }

    @Scheduled(fixedDelay = 1000)
    public void produce() {
        kafkaTemplate.send(Config.TOPIC_TEST, 0, "" + System.currentTimeMillis(), new User("张三", 1, LocalDateTime.now(), LocalDateTime.now()));
    }
}

5 User

package com.gzz.service;

import java.io.Serial;
import java.io.Serializable;
import java.time.LocalDateTime;

import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
import com.fasterxml.jackson.databind.annotation.JsonSerialize;
import com.fasterxml.jackson.datatype.jsr310.deser.LocalDateTimeDeserializer;
import com.fasterxml.jackson.datatype.jsr310.ser.LocalDateTimeSerializer;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
/**
 * @author https://www.jianshu.com/u/3bd57d5f1074
 * @date 2023-02-27 14:50:00
 */
@Data
@AllArgsConstructor
@NoArgsConstructor
public class User implements Serializable {
    @Serial
    private static final long serialVersionUID = 1L;
    private String name;
    private Integer age;
    @JsonSerialize(using = LocalDateTimeSerializer.class) // 序列化
    @JsonDeserialize(using = LocalDateTimeDeserializer.class) // 反序列化
    private LocalDateTime birthday;
    @JsonDeserialize(using = LocalDateTimeDeserializer.class) // 反序列化
    @JsonSerialize(using = LocalDateTimeSerializer.class) // 序列化
    private LocalDateTime enterDate;
}

6 Application

package com.gzz;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;

/**
 * @author https://www.jianshu.com/u/3bd57d5f1074
 * @date 2023-02-27 14:50:00
 */
@EnableScheduling
@SpringBootApplication
public class Application {
    public static void main(String args[]) {
        SpringApplication.run(Application.class);
    }
}

7 logback.xml

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE xml>
<configuration>
    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <pattern>%d[%F:%L][%p]:%m%n</pattern>
        </encoder>
    </appender>
    <logger name="org.apache.kafka" level="ERROR" />
    <root level="info">
        <appender-ref ref="STDOUT" />
    </root>
</configuration>

1677500183941.png