spring整合kafka如何实现消息路由

在Spring中整合Kafka并实现消息路由，你需要遵循以下步骤：

1. 添加依赖

在你的项目中，添加Spring Kafka和Kafka客户端的依赖。在Maven项目的pom.xml文件中添加以下依赖：

    org.springframework.kafka
    spring-kafka
    2.7.4


    org.apache.kafka
    kafka-clients
    2.8.0

2. 配置Kafka

在application.properties或application.yml文件中配置Kafka相关的属性，例如：

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=my-group
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

3. 创建Kafka消息监听器

创建一个Kafka消息监听器类，用于处理接收到的消息。这个类需要实现ConsumerAwareErrorHandler接口，以便在发生错误时进行处理。

@Service
public class KafkaMessageListener {

    @KafkaListener(topics = "${kafka.topic}", groupId = "${kafka.consumer.group-id}")
    public void listen(ConsumerRecord record, Consumer consumer) {
        // 处理接收到的消息
        System.out.printf("Received message: key = %s, value = https://www.yisu.com/ask/%s, partition = %d, offset = %d%n",
                record.key(), record.value(), record.partition(), record.offset());
    }

    @Override
    public void handle(Exception thrownException, ConsumerRecord data, Consumer consumer) {
        // 处理错误
        System.err.printf("Error occurred while processing record: %s%n", thrownException.getMessage());
    }
}

4. 创建Kafka消息生产者

创建一个Kafka消息生产者类，用于发送消息到Kafka。

@Service
public class KafkaMessageProducer {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    public void sendMessage(String topic, String message) {
        kafkaTemplate.send(topic, message);
    }
}

5. 实现消息路由逻辑

根据你的业务需求，实现消息路由逻辑。例如，你可以根据消息的内容将消息发送到不同的主题。

@Service
public class MessageRouter {

    @Autowired
    private KafkaMessageProducer kafkaMessageProducer;

    public void routeMessage(String message) {
        // 根据消息内容决定发送到哪个主题
        if (message.contains("topic1")) {
            kafkaMessageProducer.sendMessage("topic1", message);
        } else if (message.contains("topic2")) {
            kafkaMessageProducer.sendMessage("topic2", message);
        } else {
            // 默认主题
            kafkaMessageProducer.sendMessage("default-topic", message);
        }
    }
}

6. 使用消息路由

在你的应用程序中，使用MessageRouter类来处理消息并实现消息路由。

@RestController
public class MessageController {

    @Autowired
    private MessageRouter messageRouter;

    @PostMapping("/route")
    public ResponseEntity routeMessage(@RequestBody String message) {
        messageRouter.routeMessage(message);
        return ResponseEntity.ok("Message routed successfully");
    }
}

现在，当你的应用程序接收到一个消息时，MessageRouter类将根据消息的内容将其路由到相应的主题。