SpringBoot集成kafka-监听器注解
SpringBoot集成kafka-监听器注解
- 1、application.yml
- 2、生产者
- 3、消费者
- 4、测试类
- 5、测试
1、application.yml
#自定义配置
kafka:topic:name: helloTopicconsumer:group: helloGroup
2、生产者
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 {@Resourceprivate 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);}}
3、消费者
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 = {"${kafka.topic.name}"},groupId="${kafka.consumer.group}")public void onEvent3(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());}}
4、测试类
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 {@Resourceprivate EventProducer eventProducer;@Testvoid sendEvent2(){eventProducer.sendEvent2();}}
5、测试
先启动消费者监听
再启动生产者发送消息
消费者依然可以正常监听到消息：
