听起来您正在请求类似于我们在 Spring AMQP 测试框架中提供的内容:https://docs.spring.io/spring-amqp/docs/2.0.3.RELEASE/reference/html/_reference.html#test-harness

Sounds like you are requesting something similar what we have in Spring AMQP Testing Framework: https://docs.spring.io/spring-amqp/docs/2.0.3.RELEASE/reference/html/_reference.html#test-harness

所以，如果你不擅长额外的变量，你可以借用 解决方案并实现您自己的线束".

So, if you are not good with extra variable you can borrow that solution and implement your own "harness".

我认为这应该是对框架的一个很好的补充，所以，请提出一个适当的issue，我们可以一起为公众带来这样的工具.

I think that should be a good addition to the Framework so, please, raise an appropriate issue and we can together bring such a tool for the public.

更新

因此，根据 Spring AMQP 基础，我在我的测试配置中这样做了:

So, according Spring AMQP foundation I did this in my test configuration:

public static class KafkaListenerTestHarness extends KafkaListenerAnnotationBeanPostProcessor {

    private final Map<String, Object> listeners = new HashMap<>();

    @Override
    protected void processListener(MethodKafkaListenerEndpoint endpoint, KafkaListener kafkaListener,
            Object bean, Object adminTarget, String beanName) {

        bean = Mockito.spy(bean);

        this.listeners.put(kafkaListener.id(), bean);

        super.processListener(endpoint, kafkaListener, bean, adminTarget, beanName);
    }

    @SuppressWarnings("unchecked")
    public <T> T getSpy(String id) {
        return (T) this.listeners.get(id);
    }

}

...

@SuppressWarnings("rawtypes")
@Bean(name = KafkaListenerConfigUtils.KAFKA_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public static KafkaListenerTestHarness kafkaListenerAnnotationBeanPostProcessor() {
    return new KafkaListenerTestHarness();
}

然后在目标测​​试用例中我像这样使用它:

Then in the target test-case I use it like this:

@Autowired
private KafkaListenerTestHarness harness;
...
Listener listener = this.harness.getSpy("foo");

verify(listener, times(2)).listen1("foo");