自动化测试(十一) 事件驱动测试-Kafka-RabbitMQ消息组件测试

事件驱动测试：Kafka/RabbitMQ消息组件测试

微服务架构下，服务间通信不只是HTTP调用，消息队列（Kafka/RabbitMQ）也是重要一环。今天咱们聊聊怎么测试基于消息的系统——这比接口测试更复杂，因为引入了异步和最终一致性。

一、消息系统的测试挑战

假设订单服务创建订单后，发送一个OrderCreatedEvent到Kafka，库存服务消费这个消息来扣减库存：

┌─────────────┐ OrderCreatedEvent ┌─────────────┐ │ 订单服务 │ ────────────────────────> │ 库存服务 │ │ order-svc │ (Kafka) │ stock-svc │ └─────────────┘ └─────────────┘

测试难点

二、Kafka 测试方案

方案1：嵌入式 Kafka（Spring Kafka Test）

适合单元测试和集成测试，不需要Docker。

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

@SpringBootTest@EmbeddedKafka(partitions=1,// 单分区，保证顺序topics={"order-events"},// 自动创建topicbrokerProperties={"listeners=PLAINTEXT://localhost:9092","port=9092"})classOrderEventProducerTest{@AutowiredOrderEventProducerproducer;@AutowiredEmbeddedKafkaBrokerembeddedKafka;@Test@DisplayName("订单创建后发送事件到Kafka")voidshouldSendOrderCreatedEvent(){// GivenOrderCreatedEventevent=newOrderCreatedEvent(100L,"ITEM-001",2,1L);// Whenproducer.send(event);// Then: 消费验证Consumer<String,String>consumer=createConsumer();embeddedKafka.consumeFromAnEmbeddedTopic(consumer,"order-events");ConsumerRecord<String,String>record=KafkaTestUtils.getSingleRecord(consumer,"order-events",Duration.ofSeconds(5));assertThat(record).isNotNull();assertThat(record.key()).isEqualTo("100");OrderCreatedEventreceived=objectMapper.readValue(record.value(),OrderCreatedEvent.class);assertThat(received.getOrderId()).isEqualTo(100L);assertThat(received.getSku()).isEqualTo("ITEM-001");consumer.close();}privateConsumer<String,String>createConsumer(){Map<String,Object>props=KafkaTestUtils.consumerProps("test-group","true",embeddedKafka);props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");returnnewDefaultKafkaConsumerFactory<>(props,newStringDeserializer(),newStringDeserializer()).createConsumer();}}

方案2：Testcontainers + Kafka（更接近真实环境）

@SpringBootTest@TestcontainersclassOrderEventIntegrationTest{@ContainerstaticKafkaContainerkafka=newKafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.5.0"));@DynamicPropertySourcestaticvoidconfigureKafka(BootstrapRegistryregistry){registry.add("spring.kafka.bootstrap-servers",kafka::getBootstrapServers);}@AutowiredOrderServiceorderService;@AutowiredKafkaTemplate<String,String>kafkaTemplate;@Test@DisplayName("创建订单触发库存扣减（端到端）")voidshouldDeductStockWhenOrderCreated()throwsException{// Given: 准备一个消费者来验证库存服务收到的消息Consumer<String,String>consumer=createConsumer("stock-service-group");consumer.subscribe(List.of("order-events"));// When: 创建订单OrderResultresult=orderService.createOrder(newCreateOrderRequest("ITEM-001",2));assertThat(result.isSuccess()).isTrue();// Then: 验证消息被发送（最多等5秒）ConsumerRecords<String,String>records=consumer.poll(Duration.ofSeconds(5));assertThat(records.count()).isEqualTo(1);ConsumerRecord<String,String>record=records.iterator().next();OrderCreatedEventevent=objectMapper.readValue(record.value(),OrderCreatedEvent.class);assertThat(event.getOrderId()).isEqualTo(result.getOrderId());assertThat(event.getEventType()).isEqualTo("ORDER_CREATED");consumer.close();}}

方案3：Kafka 消费者测试

@SpringBootTest@EmbeddedKafka(topics="order-events")classStockEventConsumerTest{@AutowiredStockEventConsumerconsumer;@AutowiredStockServicestockService;@SpyBean// 部分Mock，监控调用StockEventConsumerconsumerSpy;@Test@DisplayName("消费订单创建事件，扣减库存")voidshouldDeductStockOnOrderCreated(){// GivenOrderCreatedEventevent=newOrderCreatedEvent(100L,"ITEM-001",2,1L);ConsumerRecord<String,String>record=newConsumerRecord<>("order-events",0,0L,"100",toJson(event));// When: 直接调用消费者方法（模拟Kafka投递）consumer.onOrderCreated(record);// Thenverify(stockService).deduct("ITEM-001",2);}@Test@DisplayName("重复消费同一事件，只扣减一次库存（幂等性）")voidshouldBeIdempotent(){OrderCreatedEventevent=newOrderCreatedEvent(100L,"ITEM-001",2,1L);ConsumerRecord<String,String>record=newConsumerRecord<>("order-events",0,0L,"100",toJson(event));// 消费两次（模拟重复投递）consumer.onOrderCreated(record);consumer.onOrderCreated(record);// 只扣减一次verify(stockService,times(1)).deduct("ITEM-001",2);}@Test@DisplayName("库存不足时，消息进入死信队列")voidshouldSendToDLQWhenInsufficientStock(){when(stockService.deduct(any(),any())).thenThrow(newInsufficientStockException());OrderCreatedEventevent=newOrderCreatedEvent(100L,"ITEM-001",999,1L);ConsumerRecord<String,String>record=newConsumerRecord<>("order-events",0,0L,"100",toJson(event));// 消费失败assertThatThrownBy(()->consumer.onOrderCreated(record)).isInstanceOf(InsufficientStockException.class);// 验证重试后进入死信队列// ... 验证DLQ topic收到消息}}

三、RabbitMQ 测试方案

Testcontainers + RabbitMQ

<dependency><groupId>org.testcontainers</groupId><artifactId>rabbitmq</artifactId><scope>test</scope></dependency>

@SpringBootTest@TestcontainersclassRabbitMQOrderEventTest{@ContainerstaticRabbitMQContainerrabbitMQ=newRabbitMQContainer(DockerImageName.parse("rabbitmq:3-alpine"));@DynamicPropertySourcestaticvoidconfigureProperties(DynamicPropertyRegistryregistry){registry.add("spring.rabbitmq.host",rabbitMQ::getHost);registry.add("spring.rabbitmq.port",rabbitMQ::getAmqpPort);registry.add("spring.rabbitmq.username",rabbitMQ::getAdminUsername);registry.add("spring.rabbitmq.password",rabbitMQ::getAdminPassword);}@AutowiredRabbitTemplaterabbitTemplate;@AutowiredOrderEventPublisherpublisher;@Test@DisplayName("发送订单事件到RabbitMQ")voidshouldPublishOrderEvent(){// GivenOrderCreatedEventevent=newOrderCreatedEvent(100L,"ITEM-001",2,1L);// Whenpublisher.publish(event);// Then: 直接从队列取消息验证Messagemessage=rabbitTemplate.receive("order-events-queue",5000);assertThat(message).isNotNull();OrderCreatedEventreceived=objectMapper.readValue(newString(message.getBody()),OrderCreatedEvent.class);assertThat(received.getOrderId()).isEqualTo(100L);}}

RabbitMQ 消费者测试

@SpringBootTestclassStockRabbitConsumerTest{@AutowiredStockRabbitConsumerconsumer;@AutowiredStockServicestockService;@Test@DisplayName("消费订单消息，扣减库存")voidshouldConsumeAndDeduct(){OrderCreatedEventevent=newOrderCreatedEvent(100L,"ITEM-001",2,1L);Messagemessage=MessageBuilder.withBody(toJson(event).getBytes()).setContentType(MessageProperties.CONTENT_TYPE_JSON).build();// 直接调用@RabbitListener方法consumer.handleOrderCreated(message);verify(stockService).deduct("ITEM-001",2);}}

四、异步等待策略

消息测试最头疼的是"什么时候断言"。

方案1：Awaitility（推荐）

<dependency><groupId>org.awaitility</groupId><artifactId>awaitility</artifactId><version>4.2.0</version><scope>test</scope></dependency>

@Test@DisplayName("订单创建后，库存最终会被扣减")voidshouldEventuallyDeductStock(){// 创建订单orderService.createOrder(newCreateOrderRequest("ITEM-001",2));// 等待最多5秒，直到库存被扣减await().atMost(Duration.ofSeconds(5)).pollInterval(Duration.ofMillis(200)).untilAsserted(()->{Stockstock=stockRepository.findBySku("ITEM-001");assertThat(stock.getAvailableQuantity()).isEqualTo(98);// 原100-2});}

方案2：CountDownLatch（精确控制）

@Test@DisplayName("批量订单并发创建，消息顺序消费")voidshouldConsumeInOrder()throwsInterruptedException{intorderCount=10;CountDownLatchlatch=newCountDownLatch(orderCount);List<Long>receivedOrderIds=Collections.synchronizedList(newArrayList<>());// 设置消费者回调consumer.setMessageHandler(event->{receivedOrderIds.add(event.getOrderId());latch.countDown();});// 发送10个订单for(inti=0;i<orderCount;i++){orderService.createOrder(newCreateOrderRequest("ITEM-001",1));}// 等待所有消息被消费booleanallConsumed=latch.await(10,TimeUnit.SECONDS);assertThat(allConsumed).isTrue();// 验证顺序assertThat(receivedOrderIds).hasSize(orderCount);assertThat(receivedOrderIds).isSorted();// 按顺序消费}

五、消息测试的黄金法则

六、小结

今天咱们聊了事件驱动系统的测试：

一句话总结：消息测试的核心是"异步等待"——用Awaitility等工具优雅处理最终一致性，同时别忘了验证幂等性和死信处理。