java - 一个 bean 可以依赖另一个 bean 仅用于 Spring 中的测试吗？

Question

我使用Spring进行依赖注入(inject)，并且我有一个bean，它恰好是Kafka生产者服务，它通过属性文件获取它的配置，如zookeeper服务器等。

import kafka.utils.ZKStringSerializer$;
import kafka.utils.ZkUtils;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.ZkConnection;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

@Service
public class KafkaProducerService implements InitializingBean {

    @Autowired
    private Properties properties;

    private KafkaProducer<String, String> producer;
    private ZkUtils zkUtils;

    public KafkaProducerService() {
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        Properties kafkaProducerProperties = new Properties();
        kafkaProducerProperties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        kafkaProducerProperties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        kafkaProducerProperties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getProperty("kafka.bootstrap.servers"));
        kafkaProducerProperties.setProperty(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, properties.getProperty("kafka.producer.timeout", "3000"));
        kafkaProducerProperties.setProperty(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG, properties.getProperty("kafka.reconnect.backoff.ms", "1000"));
        kafkaProducerProperties.setProperty(ProducerConfig.MAX_BLOCK_MS_CONFIG, properties.getProperty("kafka.producer.timeout", "3000"));
        kafkaProducerProperties.setProperty(ProducerConfig.ACKS_CONFIG, "1");
        String zookeeperEndpoint = properties.get("zookeeper.connect") + ":2181";
        this.producer = new KafkaProducer<>(kafkaProducerProperties);
        final ZkClient zkClient = new ZkClient(zookeeperEndpoint, 10000, 10000, ZKStringSerializer$.MODULE$);
        zkUtils = new ZkUtils(zkClient, new ZkConnection(zookeeperEndpoint), false);
    }

    public boolean publishMessage(final String message, final String topic) {
        try {
            producer.send(new ProducerRecord<>(topic, message))
                    .get(3, TimeUnit.SECONDS);
            return true;
        } catch (InterruptedException | ExecutionException | TimeoutException e) {
            return false;
        }
    }

    public void tearDown() {
        this.producer.close();
    }
}

我通过将其 Autowiring 到其他服务来使用此服务，并且在运行应用程序并使用它时效果很好。我有额外的 spring 上下文用于加载嵌入式 Kafka 的测试。

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:tx="http://www.springframework.org/schema/tx"
       xmlns:context="http://www.springframework.org/schema/context"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.1.xsd
        http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.1.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.0.xsd">

    <context:annotation-config/>
    <context:component-scan base-package="my.project.main"/>

    <context:property-placeholder location="classpath*:my_properties.properties"/>

    <bean id="embeddedKafka" class="my.project.main.EmbeddedKafka"
          init-method="setupEmbeddedKafkaWithZookeeper"
          destroy-method="tearDown"/>

    <bean id="properties" class="org.springframework.beans.factory.config.PropertiesFactoryBean">
        <property name="properties">
            <props>
                <prop key="zookeeper.connect">localhost</prop>
                <prop key="kafka.bootstrap.servers">localhost:9092</prop>
                <prop key="acks">1</prop>
                <prop key="kafka.producer.timeout">5000</prop>
                <prop key="kafka.reconnect.backoff.ms">30000</prop>
            </props>
        </property>
    </bean>
</beans>

每当我运行测试时，kafka 服务生产者都应该使用通过 EmbeddedKafka bean 加载的内存中的 kafka。问题在于，大多数情况下，嵌入式kafka启动时间过长，导致kafka生产者超时，无法被Spring实例化。是否有任何机制可以使 KafkaProducerService bean“等待”直到 EmbeddedKafka bean 实例化？

Answer 1

有一些方法可以让生产者等待。但我猜你不想永远等待。任何等待都会有一些超时(假设 X 秒)，以防万一嵌入式 kafka 配置不正确或发生其他问题，以便测试不会永远挂起。

您可以将生产者的超时设置为 X 秒，这样就可以了。

如果您确实想永远等待，请耐心等待。

您的主要目标是确保 kafka 在第一次访问 Zookeeper/kafka 之前已启动。

如果在 spring 上下文初始化期间没有发生这种情况(您可以在超时时获得的堆栈跟踪中检查这一点)，那么您唯一需要确保的是嵌入式 kafka init 是同步完成的。

例如，您可以创建新的 bean 来调用 setupEmbeddedKafkaWithZookeeper 并等待 kafka 启动。

如果在spring上下文初始化期间访问zookeeper/kafka，那就更棘手了。您要么需要

1. 等到 kafka 在发生的那一刻启动
2. 或者在创建 Spring 上下文之前初始化 kafka。

为了等待 kafka，您可以为 KafkaProducerService 创建一个包装器，它将等待 kafka 在访问 kafka/zookeeper 的所有方法中启动。

或者，您可以通过创建自己的继承(或包装)运行器来初始化嵌入式kafkaspring runner这将在 Spring 上下文创建之前进行初始化。