如何在spring-boot应用程序中使用@KafkaListener注释测试方法？

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class KafkaConsumer {

    @KafkaListener(topics = "my-topic")
    public void consume(String message) {
        // 处理接收到的消息
        System.out.println("Received message: " + message);
    }
}

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.kafka.test.context.EmbeddedKafkaBroker;
import org.springframework.kafka.test.utils.KafkaTestUtils;

@SpringBootTest
@EmbeddedKafka(topics = "my-topic", partitions = 1)
public class KafkaConsumerTest {

    @Autowired
    private EmbeddedKafkaBroker embeddedKafkaBroker;

    @Autowired
    private KafkaConsumer kafkaConsumer;

    @Test
    public void testConsume() {
        // 发送测试消息到Kafka主题
        String message = "Test message";
        KafkaTestUtils.getProducerProperties(embeddedKafkaBroker)
                .put("bootstrap.servers", embeddedKafkaBroker.getBrokersAsString());
        KafkaTestUtils.produceSingleRecord(embeddedKafkaBroker, "my-topic", message);

        // 等待消费者处理消息
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        // 验证消费者是否成功接收到消息
        // 可以使用断言或其他验证方式
    }
}