MongoDB - 如何在Spring重新选举期间处理写入失败？

Question

我用Spring配置了我的MongoDB副本集,我正在尝试测试自动故障转移.我知道如果主服务器出现故障,则需要几秒钟才能选出新的主服务器,因此在该时间段内,所有写操作都将失败.

我有一个测试应用程序每1秒写入一次数据库,当我取下主数据库时,我得到一个java.io.IOException(因为没有要写入的主数据库).如果我重新启动我的应用程序,则执行写操作而不会出现问题.

我认为MongoDB Java驱动程序可以使用重试来处理这些情况(我错了吗？),但我无法配置Spring来做到这一点,所以我会提供一些帮助.:)

我的配置是这样的:

<mongo:mongo id="mongo" replica-set="host1:27017,host2:27017,host3:27017">
    <mongo:options
        connections-per-host="8"
        threads-allowed-to-block-for-connection-multiplier="4"
        connect-timeout="1000"
        max-wait-time="1500"
        auto-connect-retry="true"
        socket-keep-alive="true"
        socket-timeout="1500"
        slave-ok="true"
        write-number="1"
        write-timeout="0"
        write-fsync="true"/>
</mongo:mongo>

<mongo:repositories base-package="my.repositories" />

<mongo:db-factory dbname="my_db" mongo-ref="mongo" />

<bean id="mongoTemplate" class="org.springframework.data.mongodb.core.MongoTemplate">
    <constructor-arg name="mongoDbFactory" ref="mongoDbFactory" />
</bean>

谢谢!

Answer 1

这是一个弹簧aop/spring重试自定义RetryPolicy的初始刺,用于在各种情况下进行通用重试.这非常脆弱(因为它使用了可能发生变化的异常消息等).我建议进行强大的测试,并且肯定会重复更改MongoDB和/或java驱动程序版本.

首先,maven依赖使用:

    <dependencies>
    <dependency>
        <groupId>org.mongodb</groupId>
        <artifactId>mongo-java-driver</artifactId>
        <version>2.11.3</version>
    </dependency>
    <dependency>
        <groupId>org.springframework.data</groupId>
        <artifactId>spring-data-mongodb</artifactId>
        <version>1.3.2.RELEASE</version>
    </dependency>
    <dependency>
        <groupId>org.aspectj</groupId>
        <artifactId>aspectjweaver</artifactId>
        <version>1.6.2</version>
    </dependency>
    <dependency>
        <groupId>org.aspectj</groupId>
        <artifactId>aspectjrt</artifactId>
        <version>1.6.2</version>
    </dependency>
    <dependency>
        <groupId>org.aspectj</groupId>
        <artifactId>aspectjtools</artifactId>
        <version>1.6.2</version>
    </dependency>
    <dependency>
        <groupId>org.springframework.retry</groupId>
        <artifactId>spring-retry</artifactId>
        <version>1.0.3.RELEASE</version>
    </dependency>

</dependencies>

第二,自定义org.springframework.retry.RetryPolicy

import org.springframework.retry.RetryContext;
import org.springframework.retry.policy.SimpleRetryPolicy;

import java.util.HashMap;
import java.util.Map;
import java.util.logging.Logger;

public class CustomMongoDBRetryPolicy extends SimpleRetryPolicy {
    private static final Logger logger = Logger.getLogger(CustomMongoDBRetryPolicy.class.getName());
    public CustomMongoDBRetryPolicy(int maxAttempts) {
        super(maxAttempts, createRetryableExceptions(), true);
    }

    private static Map<Class<? extends Throwable>, Boolean> createRetryableExceptions() {
        HashMap<Class<? extends Throwable>, Boolean> classBooleanHashMap = new HashMap<Class<? extends Throwable>, Boolean>();
        classBooleanHashMap.put(org.springframework.dao.DataAccessResourceFailureException.class, true);
        classBooleanHashMap.put(org.springframework.data.mongodb.UncategorizedMongoDbException.class, true);
        classBooleanHashMap.put(com.mongodb.MongoException.class, true);
        classBooleanHashMap.put(java.net.ConnectException.class, true);
        return classBooleanHashMap;
    }

    @Override
    public boolean canRetry(RetryContext context) {
        boolean retry = super.canRetry(context);
        if (retry) {
            @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
            Throwable lastThrowable = context.getLastThrowable();
            if (lastThrowable != null) {
                String message = lastThrowable.getMessage();
                Throwable cause = lastThrowable.getCause();
                if (message != null) {
                    if (message.startsWith("No replica set members available in")) {
                        logger.info("Retrying because no replica set members available. "+message);
                        return true;
                    }
                    if (message.startsWith("not talking to master and retries used up")) {
                        logger.info("Retrying because no master. "+message);
                        return true;
                    }
                    if (message.startsWith("can't find a master")) {
                        logger.info("Retrying because no master. "+message);
                        return true;
                    }
                    if (message.matches("Read operation to server [^\\s]* failed on database .*")) {
                        logger.info("Retrying because read operation failed. "+message);
                        return true;
                    }
                }
                if (cause != null) {
                    String causeMessage = cause.getMessage();
                    if (causeMessage != null) {
                        if (causeMessage.startsWith("Connection refused")) {
                            logger.info("Retrying because connection not available. "+causeMessage+"("+message+")");
                            return true;
                        }
                    }
                }
                logger.info("Not retrying. "+message+" "+lastThrowable.getClass().getName());
                return false;
            }
        }
        return retry;
    }
}

最后,使用弹簧AOP绑定到Dao

<aop:config proxy-target-class="false">
    <aop:pointcut id="retry"
                  expression="execution(* IMyDao.count(..))" />
    <aop:pointcut id="retry2"
                  expression="execution(* IMyDao.insert(..))" />
    <aop:advisor pointcut-ref="retry"
                 advice-ref="retryAdvice" order="-1"/>
    <aop:advisor pointcut-ref="retry2"
                 advice-ref="retryAdvice" order="-1"/>
</aop:config>

以下结合org.springframework.retry.backoff.ExponentialBackOffPolicy,以延迟重试,org.springframework.retry.policy.TimeoutRetryPolicy,以限制重试时间和CustomMongoDBRetryPolicy,它重试似乎可重试的...

<bean id="retryAdvice"
      class="org.springframework.retry.interceptor.RetryOperationsInterceptor">
    <property name="retryOperations">
        <bean class="org.springframework.retry.support.RetryTemplate">
            <property name="retryPolicy">
                <bean class="org.springframework.retry.policy.CompositeRetryPolicy">
                    <property name="optimistic" value="false"/>
                    <property name="policies">
                        <set>
                            <bean class="org.springframework.retry.policy.TimeoutRetryPolicy">
                                <property name="timeout" value="20000"/>
                            </bean>
                            <bean class="CustomMongoDBRetryPolicy">
                                <constructor-arg value="100"/>
                            </bean>
                        </set>
                    </property>
                </bean>
            </property>
            <property name="listeners">
                <set>
                    <bean class="MyRetryListener"/>
                </set>
            </property>
            <property name="backOffPolicy">
                <bean class="org.springframework.retry.backoff.ExponentialBackOffPolicy">
                    <property name="initialInterval" value="500"/>
                    <property name="maxInterval" value="8000"/>
                    <property name="multiplier" value="1.5"/>
                </bean>
            </property>
        </bean>
    </property>
  </bean>

我已经用各种场景对它进行了测试,它似乎处理得最好.但是,它是否适用于特定应用,需要根据具体情况进行回答.

• 初始replicaset启动 - 服务器监听之前的常规自动连接句柄,这在主要选举之前处理 - 对应用程序都是不可见的(长期滞后)
• 杀死正在进行的主要写入操作对应用程序失败,后续重试
• 踩下主服务器,关闭主服务器 - 杀死主服务器.
• 完全复制重启(如果足够快)

希望这可以帮助