如何为Google Cloud Pub/Sub发布消息编写可测试的JUnit用例-java教程-PHP中文网

如何为google cloud pub/sub发布消息编写可测试的junit用例

在为Google Cloud Pub/Sub发布消息编写JUnit测试时，直接模拟`Publisher.Builder`及其后续方法链面临挑战，因为这些类和方法通常难以被标准测试框架拦截。本文将深入探讨这一问题，并提供一种通过引入接口和依赖注入来解耦业务逻辑与Pub/Sub客户端实现的方法，从而显著提升代码的可测试性，并演示如何使用Mockito编写有效的单元测试。

理解测试挑战：直接依赖与耦合

在提供的代码示例中，publishJSON方法直接在内部创建了Publisher实例：

Publisher publisher = Publisher.newBuilder(topicName).build();

登录后复制

这种模式导致了高度耦合：业务逻辑（publishJSON方法）与具体的Pub/Sub客户端实现（Publisher类及其构建器）紧密绑定。在单元测试中，我们通常希望隔离被测试单元，避免其与外部系统（如Google Cloud Pub/Sub服务）进行实际交互。然而，由于Publisher.newBuilder()是一个静态工厂方法，并且Publisher类本身可能包含final方法或难以模拟的内部结构，传统的模拟框架（如Mockito）在尝试直接模拟Publisher.Builder或Publisher的特定行为时会遇到困难。

这意味着，在不实际连接到Pub/Sub服务的情况下，很难验证publishJSON方法是否正确地构建了消息、调用了发布逻辑，以及处理了返回结果或异常。

解决方案：引入接口和依赖注入

为了解决上述问题，核心思想是遵循“依赖倒置原则”和“依赖注入”模式。我们将创建一个抽象接口来定义Pub/Sub发布行为，然后让我们的服务类依赖于这个接口，而不是具体的实现。

1. 定义Pub/Sub发布接口

首先，定义一个描述Pub/Sub发布操作的接口。这个接口应该足够通用，以便在测试中可以轻松地模拟其行为。

秘塔写作猫

秘塔写作猫是一个集AI写作、校对、润色、配图等为一体的创作平台

127

查看详情

// src/main/java/com/example/pubsub/PubSubMessagePublisher.java
package com.example.pubsub;

import java.io.IOException;
import java.util.concurrent.ExecutionException;

public interface PubSubMessagePublisher {
    /**
     * 发布一个JSON字符串作为Pub/Sub消息。
     * @param json 要发布的JSON字符串。
     * @return 发布成功的消息ID。
     * @throws InterruptedException 如果线程在等待消息ID时被中断。
     * @throws ExecutionException 如果异步发布操作失败。
     * @throws IOException 如果在Pub/Sub客户端操作中发生IO错误。
     */
    String publish(String json) throws InterruptedException, ExecutionException, IOException;
}

登录后复制

2. 实现Pub/Sub发布接口

接下来，创建一个具体的实现类，它将封装与Google Cloud Pub/Sub客户端库的实际交互。这个实现类将包含原始publishJSON方法中的大部分逻辑。

// src/main/java/com/example/pubsub/GoogleCloudPubSubPublisher.java
package com.example.pubsub;

import com.google.api.core.ApiFuture;
import com.google.cloud.pubsub.v1.Publisher;
import com.google.protobuf.ByteString;
import com.google.pubsub.v1.PubsubMessage;
import com.google.pubsub.v1.TopicName;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

@Component // 假设使用Spring管理Bean
public class GoogleCloudPubSubPublisher implements PubSubMessagePublisher {

    private static final Logger log = LoggerFactory.getLogger(GoogleCloudPubSubPublisher.class);

    @Value("${gcp.pubsub.project-id}")
    private String projectId;

    @Value("${gcp.pubsub.topic-id}")
    private String topicId;

    @Override
    public String publish(String json) throws InterruptedException, ExecutionException, IOException {
        log.info("Publishing payload to topic: {}", topicId);
        TopicName topicName = TopicName.of(projectId, topicId);
        Publisher publisher = null;
        try {
            publisher = Publisher.newBuilder(topicName).build();
            ByteString data = ByteString.copyFromUtf8(json);
            PubsubMessage pubsubMessage = PubsubMessage.newBuilder().setData(data).build();
            ApiFuture<String> messageIdFuture = publisher.publish(pubsubMessage);
            String messageId = messageIdFuture.get(); // 阻塞等待消息ID
            log.info("Published message ID: {}", messageId);
            return messageId;
        } catch (ExecutionException e) {
            log.error("Error while publishing message: {}", e.getMessage(), e);
            throw e;
        } catch (IOException e) {
            log.error("PubSub IO exception: {}", e.getMessage(), e);
            throw e;
        } catch (InterruptedException e) {
            log.error("Connection making exception for PubSub: {}", e.getMessage(), e);
            // 重新设置中断状态
            Thread.currentThread().interrupt();
            throw e;
        } catch (Exception e) {
            log.error("publish error: {}", e.getMessage(), e);
            throw e;
        } finally {
            if (publisher != null) {
                // 当Publisher不再需要时，关闭以释放资源
                publisher.shutdown();
                publisher.awaitTermination(1, TimeUnit.MINUTES);
            }
        }
    }
}

登录后复制

注意： 实际应用中，Publisher实例的生命周期管理（创建、关闭）可能需要更精细的设计，例如通过Spring的@Bean方法或自定义工厂来管理，以避免每次发布都创建和关闭Publisher，从而提高性能和资源利用率。对于本教程，我们保持每次发布都创建和关闭以简化示例，但请在生产环境中考虑优化。

3. 修改服务类以使用接口

现在，原始服务类（假设名为MyService）不再直接创建Publisher，而是通过构造函数注入PubSubMessagePublisher接口的实现。

// src/main/java/com/example/service/MyService.java
package com.example.service;

import com.example.pubsub.PubSubMessagePublisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.concurrent.ExecutionException;

@Service // 假设使用Spring管理Bean
public class MyService {

    private static final Logger log = LoggerFactory.getLogger(MyService.class);
    private final PubSubMessagePublisher pubSubPublisher;

    // 通过构造函数注入PubSubMessagePublisher
    public MyService(PubSubMessagePublisher pubSubPublisher) {
        this.pubSubPublisher = pubSubPublisher;
    }

    public String processAndPublish(String data) throws InterruptedException, ExecutionException, IOException {
        log.info("Processing data for publishing: {}", data);
        // 这里可以有其他业务逻辑，例如数据转换、验证等
        String messageId = pubSubPublisher.publish(data);
        log.info("Data successfully published with ID: {}", messageId);
        return messageId;
    }
}

登录后复制

编写JUnit测试用例

通过上述重构，我们现在可以轻松地为MyService编写单元测试，而无需实际连接到Google Cloud Pub/Sub。我们将使用Mockito来模拟PubSubMessagePublisher接口。

// src/test/java/com/example/service/MyServiceTest.java
package com.example.service;

import com.example.pubsub.PubSubMessagePublisher;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;

import java.io.IOException;
import java.util.concurrent.ExecutionException;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.*;

class MyServiceTest {

    @Mock
    private PubSubMessagePublisher mockPubSubPublisher;

    @InjectMocks
    private MyService myService;

    @BeforeEach
    void setUp() {
        MockitoAnnotations.openMocks(this); // 初始化模拟对象
    }

    @Test
    void testProcessAndPublish_Success() throws InterruptedException, ExecutionException, IOException {
        // 模拟PubSubMessagePublisher的行为：当调用publish方法时，返回一个固定的消息ID
        String expectedMessageId = "test-message-id-123";
        when(mockPubSubPublisher.publish(anyString())).thenReturn(expectedMessageId);

        String testJsonData = "{\"key\": \"value\"}";
        String actualMessageId = myService.processAndPublish(testJsonData);

        // 验证返回的消息ID是否正确
        assertEquals(expectedMessageId, actualMessageId);
        // 验证mockPubSubPublisher的publish方法是否被调用了一次，且参数正确
        verify(mockPubSubPublisher, times(1)).publish(testJsonData);
    }

    @Test
    void testProcessAndPublish_PubSubInterruptedException() throws InterruptedException, ExecutionException, IOException {
        // 模拟PubSubMessagePublisher抛出InterruptedException
        doThrow(new InterruptedException("Test Interruption")).when(mockPubSubPublisher).publish(anyString());

        String testJsonData = "{\"key\": \"value\"}";

        // 验证MyService是否正确地重新抛出了异常
        assertThrows(InterruptedException.class, () -> myService.processAndPublish(testJsonData));
        verify(mockPubSubPublisher, times(1)).publish(testJsonData);
    }

    @Test
    void testProcessAndPublish_PubSubExecutionException() throws InterruptedException, ExecutionException, IOException {
        // 模拟PubSubMessagePublisher抛出ExecutionException
        doThrow(new ExecutionException("Test Execution Error", new RuntimeException())).when(mockPubSubPublisher).publish(anyString());

        String testJsonData = "{\"key\": \"value\"}";

        // 验证MyService是否正确地重新抛出了异常
        assertThrows(ExecutionException.class, () -> myService.processAndPublish(testJsonData));
        verify(mockPubSubPublisher, times(1)).publish(testJsonData);
    }

    @Test
    void testProcessAndPublish_PubSubIOException() throws InterruptedException, ExecutionException, IOException {
        // 模拟PubSubMessagePublisher抛出IOException
        doThrow(new IOException("Test IO Error")).when(mockPubSubPublisher).publish(anyString());

        String testJsonData = "{\"key\": \"value\"}";

        // 验证MyService是否正确地重新抛出了异常
        assertThrows(IOException.class, () -> myService.processAndPublish(testJsonData));
        verify(mockPubSubPublisher, times(1)).publish(testJsonData);
    }
}

登录后复制

注意事项与总结

依赖注入框架： 上述示例使用了Spring框架的@Component、@Service和@Value注解来简化依赖管理和配置注入。如果您的项目不使用Spring，您需要手动管理这些依赖的创建和注入。
Publisher生命周期： 在GoogleCloudPubSubPublisher中，我们每次调用publish都创建并关闭了Publisher。在生产环境中，这可能不是最优解。一个更好的实践是，如果GoogleCloudPubSubPublisher是单例的，那么Publisher实例可以在类初始化时创建，并在应用关闭时通过@PreDestroy或其他机制进行关闭。
异步发布： 原始代码通过messageIdFuture.get()将异步发布操作同步化。如果您的业务场景需要处理真正的异步发布，那么PubSubMessagePublisher接口可能需要返回ApiFuture<String>或CompletableFuture<String>，并且测试用例也需要相应地处理异步结果。
异常处理： 在测试中，务必验证服务层是否正确地捕获并重新抛出（或处理）底层依赖可能抛出的各种异常。
测试粒度： 这种方法允许您对MyService进行纯粹的单元测试，隔离了Pub/Sub的实际交互。对于GoogleCloudPubSubPublisher的实现，您可能需要编写集成测试，以确保它能正确地与真实的Pub/Sub服务进行通信。

通过引入接口和依赖注入，我们成功地解耦了业务逻辑与外部服务客户端，使得核心业务逻辑（MyService）变得高度可测试。这种设计模式不仅有助于编写有效的单元测试，还能提高代码的模块化、可维护性和灵活性。

以上就是如何为Google Cloud Pub/Sub发布消息编写可测试的JUnit用例的详细内容，更多请关注php中文网其它相关文章！