SpringBoot集成kafka-消息转发@sendTo()注解

1、消费者

• 启动消费者进行消息监听，消费者A监听到生产者发送的消息
• 使用@sendTo()注解将消息转发给消费者B

package com.power.consumer;import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Component;@Component
public class EventConsumer {@KafkaListener(topics = {"topicA"}, groupId = "aGroup")@SendTo("topicB")public String onEventA(ConsumerRecord<String, String> record) {System.out.println("消费者A接收消息,转发到消费者B record = " + record);return record.value() + "--forward message";}@KafkaListener(topics = {"topicB"}, groupId = "bGroup")public void onEventB(ConsumerRecord<String, String> record) {System.out.println("消费B消息接收到消费者A转发的消息 record = " + record);}}

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 sendEvent(){User user = User.builder().id(1).phone("15676767673").birthday(new Date()).build();String userJson = JSONUtils.toJSON(user);kafkaTemplate.send("topicA","k", userJson);}}

3、实体类对象

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;}

4、JSON工具类

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);}}
}

5、配置文件application.yml

spring:application:#应用名称name: spring-boot-05-kafka-MsgRedirect#kafka连接地址（ip+port）kafka:bootstrap-servers: <你的kafak服务器IP>:9092#配置消费者的反序列化consumer:key-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializer

6、测试类

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 SpringBoot05KafkaBaseApplication {@Resourceprivate EventProducer eventProducer;@Testvoid sendInterceptor(){eventProducer.sendEvent();}}

7、测试

• 先启动消费者进行消息监听
• 再启动生产者发送消息

消费者A先接收到消息：

消费者B后接收到消费者A转发的消息：