当前位置：首页 > article >正文

SpringBoot集成kafka接收对象消息

article 2025/3/6 11:14:26

SpringBoot集成kafka接收对象消息

1、生产者
2、消费者
3、工具类
4、消息实体对象
5、配置文件
6、启动类
7、测试类
8、测试结果

在这里插入图片描述

1、生产者

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 {

    @Resource
    private KafkaTemplate<String,Object> kafkaTemplate;

    public void sendEvent2(){
        User user = User.builder().id(10001).phone("15676767676").birthday(new Date()).build();
        String userJson = JSONUtils.toJSON(user);
        kafkaTemplate.send("helloTopic",userJson);
    }

}

2、消费者

package com.power.consumer;

import com.power.model.User;
import com.power.util.JSONUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.util.function.Consumer;

@Component
public class EventConsumer {

    //采用监听的方式接收事件（消息，数据）
    @KafkaListener(topics = {"helloTopic"},groupId="helloGroup")
        public void onEvent(String userJson,
                        @Header(value=KafkaHeaders.RECEIVED_TOPIC) String topic,
                        @Header(value=KafkaHeaders.RECEIVED_PARTITION_ID) String partition,
                        ConsumerRecord<String,String> record){
        User user =JSONUtils.toBean(userJson,User.class);
        System.out.println("读取/消费到的事件,user："+user+",topic："+topic+",partition："+partition);
        System.out.println("读取/消费到的事件："+record.toString());

    }

}

3、工具类

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、消息实体对象

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;

}

5、配置文件

spring:
  application:
    #应用名称
    name: spring-boot-02-kafka-base

  #kafka连接地址（ip+port）
  kafka:
    bootstrap-servers: <你的kafka服务器IP>:9092

6、启动类

package com.power;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class KafkaApplication {

	public static void main(String[] args) {
		SpringApplication.run(KafkaApplication.class, args);
		System.out.println("启动成功--------------------------");
	}
}

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 SpringBoot02KafkaBaseApplication {

    @Resource
    private EventProducer eventProducer;

    @Test
    void sendEvent2(){
        eventProducer.sendEvent2();
    }

}