SpringBoot 应用中的@RabbitListener 方法测试

Posted 2023-02-26

【中文标题】SpringBoot 应用中的@RabbitListener 方法测试

【英文标题】：@RabbitListener method testing in SpringBoot app

【发布时间】：2016-03-03 20:00:00

【问题描述】：

代码：

RabbitMQ 监听器：

@Component
public class ServerThroughRabbitMQ implements ServerThroughAMQPBroker 
    private static final AtomicLong ID_COUNTER=new AtomicLong();
    private final long instanceId=ID_COUNTER.incrementAndGet();


    @Autowired
    public ServerThroughRabbitMQ( UserService userService,LoginService loginService....)
....
    

    @Override
    @RabbitListener(queues = "#registerQueue.name")
    public String registerUserAndLogin(String json) 
       .....
    

服务器配置：

@Configuration
public class ServerConfig 
    @Value("$amqp.broker.exchange-name")
    private String exchangeName;
    @Value("$amqp.broker.host")
    private String ampqBrokerHost;
    @Value("$amqp.broker.quidco.queue.postfix")
    private String quidcoQueuePostfix;
    @Value("$amqp.broker.quidco.queue.durability:true")
    private boolean quidcoQueueDurability;
    @Value("$amqp.broker.quidco.queue.autodelete:false")
    private boolean quidcoQueueAutodelete;

    private String registerAndLoginQuequName;


    @PostConstruct
    public void init() 
        registerAndLoginQuequName = REGISTER_AND_LOGIN_ROUTING_KEY + quidcoQueuePostfix;
    public String getRegisterAndLoginQueueName() 
        return registerAndLoginQuequName;
    

    public String getLoginAndCheckBonusQueueName() 
        return loginAndCheckBonusQuequName;
    



    @Bean
    public ConnectionFactory connectionFactory() 
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(ampqBrokerHost);
        return connectionFactory;
    

    @Bean
    public AmqpAdmin amqpAdmin() 
        return new RabbitAdmin(connectionFactory());
    

    @Bean
    public TopicExchange topic() 
        return new TopicExchange(exchangeName);
    

    @Bean(name = "registerQueue")
    public Queue registerQueue() 
        return new Queue(registerAndLoginQuequName, quidcoQueueDurability, false, quidcoQueueAutodelete);
    


    @Bean
    public Binding bindingRegisterAndLogin() 
        return BindingBuilder.bind(registerQueue()).to(topic()).with(REGISTER_AND_LOGIN_ROUTING_KEY);
    

   

测试配置：

@EnableRabbit
@TestPropertySource("classpath:test.properties")
public class ServerThroughAMQPBrokerRabbitMQIntegrationTestConfig 
    private final ExecutorService=Executors.newCachedThreadPool();
    private LoginService loginServiceMock=mock(LoginService.class);
    private UserService userServiceMock =mock(UserService.class);

    @Bean
    public ExecutorService executor() 
        return executorService;
    

    @Bean
    public LoginService getLoginServiceMock() 
        return loginServiceMock;
    

    @Bean
    public UserService getUserService() 
        return userServiceMock;
    

    @Bean
    @Autowired
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) 
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMaxConcurrentConsumers(5);
        return factory;
    

    @Bean
    @Autowired
    public RabbitTemplate getRabbitTemplate(ConnectionFactory connectionFactory) 
        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        return rabbitTemplate;
    

    @Bean
    public ServerThroughRabbitMQ getServerThroughRabbitMQ() 
        return new ServerThroughRabbitMQ(userServiceMock, loginServiceMock,...);
    

集成测试：

@RunWith(SpringJUnit4ClassRunner.class)
@SpringApplicationConfiguration(classes =ServerConfig.class,ServerThroughAMQPBrokerRabbitMQIntegrationTestConfig.class)
@Category(IntegrationTest.class)
@TestPropertySource("classpath:test.properties")
public class ServerThroughAMQPBrokerRabbitMQIntegrationTest 
    final private ObjectMapper jackson = new ObjectMapper();
    @Autowired
    private ExecutorService executor;

    @Autowired
    private ServerThroughRabbitMQ serverThroughRabbitMQ;

    @Autowired
    private RabbitTemplate template;

    @Autowired
    private TopicExchange exchange;

    @Autowired
    UserService userService;

    @Autowired
    LoginService loginService;

    @Autowired
    private AmqpAdmin amqpAdmin;

    @Autowired
    private ServerConfig serverConfig;

    final String username = "username";
    final String email = "[e-mail]";
    final Integer tcVersion=1;
    final int quidcoUserId = 1;
    final String jwt = ProcessLauncherForJwtphpBuilderUnitWithCxtTest.EXPECTED_JWT;


    @Before
    public void cleanAfterOthersForMyself() 
        cleanTestQueues();
    

    @After
    public void cleanAfterMyselfForOthers() 
        cleanTestQueues();
    

    private void cleanTestQueues() 
        amqpAdmin.purgeQueue(serverConfig.getRegisterAndLoginQueueName(), false);
    

    @Test
    @Category(SlowTest.class,IntegrationTest.class)
    public void testRegistrationAndLogin() throws TimeoutException 
        final Waiter waiter = new Waiter();

        when(userService.register(anyString(), anyString(), anyString())).thenReturn(...);
        when(loginService....()).thenReturn(...);


        executor.submit(() -> 
            final RegistrationRequest request = new RegistrationRequest(username, email,tcVersion);
            final String response;
            try 
                //@todo: converter to convert RegistrationRequest inside next method to json
                response = (String) template.convertSendAndReceive(exchange.getName(), REGISTER_AND_LOGIN_ROUTING_KEY.toString(), jackson.writeValueAsString(request));
                waiter.assertThat(response, not(isEmptyString()));

                final RegistrationResponse registrationResponse = jackson.readValue(response, RegistrationResponse.class);
                waiter.assertThat(...);
                waiter.assertThat(...);

             catch (Exception e) 
                throw new RuntimeException(e);
            
            waiter.resume();
        );

        waiter.await(5, TimeUnit.SECONDS);
    

当我单独运行该测试时，一切正常，但是当我与其他测试一起运行它时，没有使用模拟的 ServerThroughRabbitMQ，因此一些弹簧缓存强制使用旧的兔子侦听器。

我尝试调试它，我可以看到，正确的 bean 正在自动连接到测试，但由于某种原因，旧的侦听器正在使用（旧 bean 字段 instanceId=1 新的模拟 bean instanceId=3）并且测试失败（不是确定它是怎么可能的，所以如果在现有的旧 bean 的情况下，我假设会得到一个自动装配异常）。

我尝试在 BEFORE_CLASS 之前使用 @DirtiesContext，但遇到了另一个问题（请参阅 here）

【问题讨论】：

你有机会通过 git 分享你的项目问题部分吗（通过额外的实验来检查这个测试真的很有用）？

您找到满意的解决方案了吗？

【参考方案1】：

RabbitMQ 和集成测试可能很难，因为 Rabbit MQ 保持某种状态： - 来自队列中先前测试的消息 - 来自先前测试的侦听器仍在侦听队列

有几种方法：

在开始测试之前清除所有队列（这可能就是您所说的cleanTestQueues()） 删除所有队列（或使用临时队列）并在每次测试前重新创建它们 使用 Rabbit Admin Rest API 杀死侦听器或先前测试的连接 删除虚拟主机并为每个测试重新创建基础设施（这是最残酷的方式）