Sel*_*wyn 5 spring-rabbit spring-amqp
我的问题实际上是一个后续问题
在那里它声明包装"你的监听器"并传入CountDownLatch并最终所有线程将合并.如果我们手动创建和注入消息监听器但是对于@RabbitListener注释,这个答案是有效的...我不知道如何传入CountDownLatch.该框架在幕后自动神奇地创建了消息监听器.
还有其他方法吗?
在@Gary Russell 的帮助下,我得到了答案并使用了以下解决方案。
结论:我必须承认我对这个解决方案漠不关心(感觉就像黑客),但这是我唯一可以开始工作的东西,一旦你完成了最初的一次性设置并真正理解了“工作流程”,它就不是了如此痛苦。基本上归结为定义 (2) @Beans 并将它们添加到您的集成测试配置中。
下面发布了示例解决方案并附有说明。请随时提出对此解决方案的改进建议。
1.定义一个 ProxyListenerBPP,在 spring 初始化期间将侦听指定的 clazz(即包含 @RabbitListener 的测试类)并注入下一步中定义的自定义 CountDownLatchListenerInterceptor 建议。
import org.aopalliance.aop.Advice;
import org.springframework.aop.framework.ProxyFactoryBean;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.core.Ordered;
import org.springframework.core.PriorityOrdered;
/**
* Implements BeanPostProcessor bean... during spring initialization we will
* listen for a specified clazz
* (i.e our @RabbitListener annotated class) and
* inject our custom CountDownLatchListenerInterceptor advice
* @author sjacobs
*
*/
public class ProxyListenerBPP implements BeanPostProcessor, BeanFactoryAware, Ordered, PriorityOrdered{
private BeanFactory beanFactory;
private Class<?> clazz;
public static final String ADVICE_BEAN_NAME = "wasCalled";
public ProxyListenerBPP(Class<?> clazz) {
this.clazz = clazz;
}
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = beanFactory;
}
@Override
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
return bean;
}
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
if (clazz.isAssignableFrom(bean.getClass())) {
ProxyFactoryBean pfb = new ProxyFactoryBean();
pfb.setProxyTargetClass(true); // CGLIB, false for JDK proxy (interface needed)
pfb.setTarget(bean);
pfb.addAdvice(this.beanFactory.getBean(ADVICE_BEAN_NAME, Advice.class));
return pfb.getObject();
}
else {
return bean;
}
}
@Override
public int getOrder() {
return Ordered.LOWEST_PRECEDENCE - 1000; // Just before @RabbitListener post processor
}
Run Code Online (Sandbox Code Playgroud)
2. 创建 MethodInterceptor 建议实现,它将保存对 CountDownLatch 的引用。CountDownLatch 需要在集成测试线程和@RabbitListener 的异步工作线程中引用。因此,一旦@RabbitListener 异步线程完成执行,我们就可以释放回集成测试线程。无需投票。
import java.util.concurrent.CountDownLatch;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
/**
* AOP MethodInterceptor that maps a <b>Single</b> CountDownLatch to one method and invokes
* CountDownLatch.countDown() after the method has completed execution. The motivation behind this
* is for integration testing purposes of Spring RabbitMq Async Worker threads to be able to merge
* the Integration Test thread after an Async 'worker' thread completed its task.
* @author sjacobs
*
*/
public class CountDownLatchListenerInterceptor implements MethodInterceptor {
private CountDownLatch countDownLatch = new CountDownLatch(1);
private final String methodNameToInvokeCDL ;
public CountDownLatchListenerInterceptor(String methodName) {
this.methodNameToInvokeCDL = methodName;
}
@Override
public Object invoke(MethodInvocation invocation) throws Throwable {
String methodName = invocation.getMethod().getName();
if (this.methodNameToInvokeCDL.equals(methodName) ) {
//invoke async work
Object result = invocation.proceed();
//returns us back to the 'awaiting' thread inside the integration test
this.countDownLatch.countDown();
//"reset" CountDownLatch for next @Test (if testing for more async worker)
this.countDownLatch = new CountDownLatch(1);
return result;
} else
return invocation.proceed();
}
public CountDownLatch getCountDownLatch() {
return countDownLatch;
}
}
Run Code Online (Sandbox Code Playgroud)
3. 接下来将以下@Bean添加到您的集成测试配置中
public class SomeClassThatHasRabbitListenerAnnotationsITConfig extends BaseIntegrationTestConfig {
// pass into the constructor the test Clazz that contains the @RabbitListener annotation into the constructor
@Bean
public static ProxyListenerBPP listenerProxier() { // note static
return new ProxyListenerBPP(SomeClassThatHasRabbitListenerAnnotations.class);
}
// pass the method name that will be invoked by the async thread in SomeClassThatHasRabbitListenerAnnotations.Class
// I.E the method name annotated with @RabbitListener or @RabbitHandler
// in our example 'listen' is the method name inside SomeClassThatHasRabbitListenerAnnotations.Class
@Bean(name=ProxyListenerBPP.ADVICE_BEAN_NAME)
public static Advice wasCalled() {
String methodName = "listen";
return new CountDownLatchListenerInterceptor( methodName );
}
// this is the @RabbitListener bean we are testing
@Bean
public SomeClassThatHasRabbitListenerAnnotations rabbitListener() {
return new SomeClassThatHasRabbitListenerAnnotations();
}
}
Run Code Online (Sandbox Code Playgroud)
4. 最后,在集成@Test调用中...通过rabbitTemplate发送消息触发异步线程后...现在调用从拦截器获取的CountDownLatch#await(...)方法,并确保传入一个TimeUnit 参数,以便在长时间运行的进程或出现问题时可能会超时。一旦异步集成测试线程被通知(唤醒),现在我们终于可以开始实际测试/验证/验证异步工作的结果。
@ContextConfiguration(classes={ SomeClassThatHasRabbitListenerAnnotationsITConfig.class } )
public class SomeClassThatHasRabbitListenerAnnotationsIT extends BaseIntegrationTest{
@Inject
private CountDownLatchListenerInterceptor interceptor;
@Inject
private RabbitTemplate rabbitTemplate;
@Test
public void shouldReturnBackAfterAsyncThreadIsFinished() throws Exception {
MyObject payload = new MyObject();
rabbitTemplate.convertAndSend("some.defined.work.queue", payload);
CountDownLatch cdl = interceptor.getCountDownLatch();
// wait for async thread to finish
cdl.await(10, TimeUnit.SECONDS); // IMPORTANT: set timeout args.
//Begin the actual testing of the results of the async work
// check the database?
// download a msg from another queue?
// verify email was sent...
// etc...
}
Run Code Online (Sandbox Code Playgroud)
这有点棘手@RabbitListener,但最简单的方法是向听众提供建议。
使用自定义侦听器容器工厂,只需让您的测试用例将建议添加到工厂即可。
建议是MethodInterceptor:调用将有 2 个参数;通道和(未转换的)Message. 必须在创建容器之前注入建议。
或者,使用注册表获取对容器的引用,并稍后添加建议(但您必须调用initialize()以强制应用新建议)。
另一种选择是BeanPostProcessor在将侦听器类注入容器之前简单地代理侦听器类。这样,您将在任何转换后看到方法参数;您还可以验证侦听器返回的任何结果(对于请求/回复场景)。
如果您不熟悉这些技术,我可以尝试找一些时间为您提供一个快速示例。
编辑
我发出了一个拉取请求,将示例添加到EnableRabbitIntegrationTests. 这将添加一个带有 2 个带注释的侦听器方法的侦听器 bean,该BeanPostProcessor方法在将侦听器 bean 注入到侦听器容器之前代理侦听器 bean。代理中添加了一个Advice,当收到预期消息时,该代理对锁存器进行计数。
| 归档时间: |
|
| 查看次数: |
2748 次 |
| 最近记录: |