如何在 Spring 测试中监视自动装配的 bean

这是我尝试过的并且有效的方法：

@SpringBootApplication
public class Demo1Application {

    public static void main(String[] args) {
        SpringApplication.run(Demo1Application.class,args);
    }

    @Bean
    ExecutorChAnnel kafkaErrorChAnnel(TaskExecutor taskExecutor) {
        return new ExecutorChAnnel(taskExecutor);
    }

    @Bean
    public messageHandler kafkaFailureHandler() {
        LoggingHandler handler = new LoggingHandler(LoggingHandler.Level.ERROR);
        handler.setShouldLogFullmessage(Boolean.TRUE);
        return handler;
    }

    @Bean
    IntegrationFlow kafkaFailureFlow(ExecutorChAnnel kafkaErrorChAnnel,messageHandler kafkaFailureHandler) {
        return IntegrationFlows.from(kafkaErrorChAnnel)
                .transform("payload.failedmessage")
                .handle(kafkaFailureHandler)
                .get();
    }

}

@SpringBootTest
class Demo1ApplicationTests {

    @Autowired
    ExecutorChAnnel kafkaErrorChAnnel;

    @SpyBean
    messageHandler kafkaFailureHandler;

    @Test
    void testSpyBean() throws InterruptedException {
        messagingException payload = new messageHandlingException(new Genericmessage<>("test"));
        this.kafkaErrorChAnnel.send(new Errormessage(payload));
        Thread.sleep(1000);
        mockito.verify(this.kafkaFailureHandler).handlemessage(ArgumentMatchers.any(message.class));
    }

}

也许您的问题是您没有在 LogHandlerConfiguration 配置中包含 @SpringBootTest。这就是为什么我要求一个简单的项目来玩。 您具有所有这些属性的代码太自定义了，无法复制/粘贴到我的环境中...

还要注意那个Thread.sleep(1000);。由于您的 kafkaErrorChAnnel 是 ExecutorChAnnel，消息消费发生在离开主测试线程的不同线程上，并由于竞争条件而导致失败。很难猜测正确的时间，所以最好存根一个模拟方法来实现一些线程屏障，比如 new CountDownLatch(1) 并在测试中等待它。

在主题之外，您还可以研究 Spring 集成测试框架：https://docs.spring.io/spring-integration/docs/current/reference/html/tesTing.html#test-context

所以，被挂断了为什么不 @SpyBean？有两个问题

1. 成功和失败处理程序都是 messageHandlers 混淆@SpyBean
2. Kafka 生产者等待时间太长，即 1000 毫秒

这是最终工作的结果，使用命名的 bean

@Bean("kafkaFailureHandler")
public messageHandler kafkaFailureHandler() {
    LoggingHandler handler = new LoggingHandler(LoggingHandler.Level.INFO);
    handler.setShouldLogFullmessage(Boolean.TRUE);
    return handler;
}

然后在测试中也减少了最大块

@DirtiesContext
@SpringBootTest(classes = {KafkaHandlerConfiguration.class,SwiftalkKafkaGateway.class})
@SpringIntegrationTest(noAutoStartup = {"kafkaFailureFlow"})
@TestPropertysource(properties = {
        "spring.main.bAnner-mode=off","logging.level.root=INFO","logging.level.org.springframework=INFO","logging.level.org.springframework.integration=DEBUG","spring.kafka.producer.properties.max.block.ms=50","spring.kafka.producer.bootstrap-servers=localhost:9999","spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.Stringserializer","spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.Jsonserializer",})
public class KafkaPublishFailureTest {

    private static final Logger log = LogManager.getLogger(KafkaPublishFailureTest.class);

    @Autowired
    SwiftalkKafkaGateway kafkaGateway;

    @SpyBean(name = "kafkaFailureHandler")
    messageHandler kafkaFailureHandler;

    @Test
    @SuppressWarnings("all")
    void testFailedKafkaPublish() throws InterruptedException {

        //Dummy message
        Map<String,String> map = new HashMap<>();
        map.put("key","value");
        // Publish message
        message<Map<String,String>> message = messageBuilder.withPayload(map)
                .setHeader("X-UPSTREAM-TYPE","alm")
                .setHeader("X-UPSTREAM-INSTANCE","jira")
                .setHeader("X-messaGE-KEY","key-1")
                .build();

        kafkaGateway.publish(messagE);
        verify(this.kafkaFailureHandler,timeout(500)).handlemessage(any(message.class));
    }

}

注意 spring.kafka.producer.properties.max.block.ms=50 和 verify(this.kafkaFailureHandler,timeout(500)).handlemessage(any(message.class));