查看原文
其他

用 Kafka 和 Awaitility 测试 Spring Boot 应用

ImportNew ImportNew 2021-03-11

(给ImportNew加星标,提高Java技能)

编译:ImportNew/覃佑桦

dzone.com/articles/testing-spring-boot-asynchronous-application-using



本文介绍了使用 EmbeddedKafka 和 Awaitility 测试 Spring Boot 应用的不同方法。


测试同步调用程序时,主要工作是“调用并等待”。调用指定 API 或 endpoint,然后等待响应。测试代码会阻塞主线程执行,直到 API 返回响应结果。处理完成后,把得到的响应与预期结果进行比较。


与同步程序或阻塞式调用的程序相比,异步程序的测试方式有所不同,无需阻塞主线程执行。简单地说,异步程序不会等待 API 响应。我们需要手动处理测试代码让执行保持在某个执行节点,并等待所有非阻塞操作的执行结果。在这个阶段可以使用断言。 


管理不同的线程与并发问题,编写简洁易读的单元测试并非易事。


可以通过几种方式为 Spring Boot 编写测试:用 Kafka 与 Spring Cloud Stream 的微服务连接。


让我们设计一个简单的例子。


示例


producer bean 向 Kafka topic 发送消息。

package com.techwording.scs;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
@EnableBinding(Source.class)
public class Producer {
private Source mySource;
public Producer(Source mySource) {
super();
this.mySource = mySource;
}
public Source getMysource() {
return mySource;
}
public void setMysource(Source mysource) {
mySource = mysource;
}
}


consumer bean 会监听 Kafka topic 并接收消息。


@EnableBinding(Sink.class)
public class Consumer {
private String receivedMessage;
@StreamListener(target = Sink.INPUT)
public void consume(String message) {
receivedMessage = message;
latch.countDown();
}
public String getReceivedMessage() {
return receivedMessage;
}
}


创建 topic 对应的 Kafka broker。在这个测试中,通过 spring-kafka-test 创建嵌入式 Kafka server。 

@ClassRule public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, TOPIC1);


EmbeddedKafkaRule


Spring-kafka-test 提供了一个嵌入式 Kafka broker。可以使用 JUnit @ClassRule 注解创建 Kafka broker。该规则会在测试执行之前在随机的端口上启动 Kafka 与 Zookeeper 服务器,并在测试完成后将其关闭。嵌入式 Kafka broker 无需在运行测试时提供真实的 Kafka 和 Zookeeper 实例。   


这里提供了两种实现方式:一种使用 Awaitility,另一种使用 countdown latch。


使用 Awaitility 测试


Awaitility 是一个 DSL 开发库,可以为异步 Java 程序开发 Junit 测试。可以在这里了解其官方 GitHub。下面是使用 Awaitility 实现的测试。


package com.techwording.scs;
import java.util.concurrent.TimeUnit;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.cloud.stream.test.binder.TestSupportBinderAutoConfiguration;
import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.test.context.junit4.SpringRunner;
import static org.assertj.core.api.BDDAssertions.then;
import static org.awaitility.Awaitility.waitAtMost;

@RunWith(SpringRunner.class)
@SpringBootTest(classes = { EmbeddedKafkaAwaitilityTest.App.class, EmbeddedKafkaAwaitilityTest.Producer.class, EmbeddedKafkaAwaitilityTest.Consumer.class })
@EnableBinding(Source.class)
public class EmbeddedKafkaAwaitilityTest {
@SpringBootApplication(exclude = TestSupportBinderAutoConfiguration.class)
static class App {
}
private static final String TOPIC1 = "test-topic-1";

@ClassRule
public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, TOPIC1);

@BeforeClass
public static void setup()
{
System.setProperty("spring.cloud.stream.kafka.binder.brokers",
embeddedKafka.getEmbeddedKafka()
.getBrokersAsString());
System.setProperty("spring.cloud.stream.bindings.input.destination", TOPIC1);
System.setProperty("spring.cloud.stream.bindings.input.content-type", "text/plain");
System.setProperty("spring.cloud.stream.bindings.input.group", "input-group-1");
System.setProperty("spring.cloud.stream.bindings.output.destination", TOPIC1);
System.setProperty("spring.cloud.stream.bindings.output.content-type", "text/plain");
System.setProperty("spring.cloud.stream.bindings.output.group", "output-group-1");
}

@Autowired
private Producer producer;

@Autowired
private Consumer consumer;

@Test
public void testMessageSendReceive_Awaitility()
{
producer.getMysource()
.output()
.send(MessageBuilder.withPayload("payload")
.setHeader("type", "string")
.build());
waitAtMost(5, TimeUnit.SECONDS)
.untilAsserted(() -> {
then("payload").isEqualTo(
EmbeddedKafkaAwaitilityTest.this.consumer.getReceivedMessage());
});
}
}


使用 CountDownLatch 进行测试


根据 Java 文档的描述,CountDownLatch 可以让多个线程保持等待,直到其它线程中一组操作执行完毕。要用 CountDownLatch 编写测试,首先要用 counter 初始化 latch。 


counter 值取决于测试需要等待的任务数。这里把 counter 初值设为1。一旦 producer 消息发送完毕,latch 会等待 count 变为0。consumer 负责减小 count。当 consumer 完成自己的工作,主线程将恢复运行并执行断言。  


下面是用 CountDownLatch 实现的测试:


package com.techwording.scs;
import java.util.concurrent.CountDownLatch;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.cloud.stream.test.binder.TestSupportBinderAutoConfiguration;
import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.test.context.junit4.SpringRunner;
import static org.assertj.core.api.Assertions.assertThat;

@RunWith(SpringRunner.class)
@SpringBootTest(classes = { EmbeddedKafkaLatchTest.App.class, EmbeddedKafkaLatchTest.Producer.class, EmbeddedKafkaLatchTest.Consumer.class })
@EnableBinding(Source.class)
public class EmbeddedKafkaLatchTest {
@SpringBootApplication(exclude = TestSupportBinderAutoConfiguration.class)
static class App {
}

private static final String TOPIC1 = "test-topic-1";

@ClassRule
public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, TOPIC1);

private static CountDownLatch latch = new CountDownLatch(1);

@BeforeClass
public static void setup()
{
System.setProperty("spring.cloud.stream.kafka.binder.brokers",
embeddedKafka.getEmbeddedKafka()
.getBrokersAsString());
System.setProperty("spring.cloud.stream.bindings.input.destination", TOPIC1);
System.setProperty("spring.cloud.stream.bindings.input.content-type", "text/plain");
System.setProperty("spring.cloud.stream.bindings.input.group", "input-group-1");
System.setProperty("spring.cloud.stream.bindings.output.destination", TOPIC1);
System.setProperty("spring.cloud.stream.bindings.output.content-type", "text/plain");
System.setProperty("spring.cloud.stream.bindings.output.group", "output-group-1");
}

@Autowired
private Producer producer;

@Autowired
private Consumer consumer;

@Test
public void testMessageSendReceive() throws InterruptedException
{
producer.getMysource()
.output()
.send(MessageBuilder.withPayload("payload")
.setHeader("type", "string")
.build());
latch.await();
assertThat(consumer.getReceivedMessage()).isEqualTo("payload");
}
}


在这里可以下载完整源代码。 

github.com/nakulshukla08/techwording/tree/master/spring-cloud-stream-kafka-example


推荐阅读  点击标题可跳转

大白话带你认识 Kafka

Spring Boot面试问题集锦

SpringBoot 并发登录人数控制


看完本文有收获?请转发分享给更多人

关注「ImportNew」,提升Java技能

好文章,我在看❤️

    您可能也对以下帖子感兴趣

    文章有问题?点此查看未经处理的缓存