Spring Cache 扩展：Redis 批量操作优化方案与 BatchCache 自定义实现

Spring Cache是一个强大的缓存抽象层，提供了统一的缓存操作接口，但原生支持主要集中在单键操作。在高并发场景下，批量操作能力对于提升系统性能至关重要。本文将深入探讨如何通过继承Cache接口，以RedisCache为基础，实现兼容Spring Cache规范的BatchCache扩展。

一、Spring Cache架构与RedisCache源码剖析

1.1 Spring Cache核心组件

Spring Cache的核心是org.springframework.cache.Cache接口，定义了缓存的基本操作：


public interface Cache {
    String getName();
    Object getNativeCache();
    ValueWrapper get(Object key);
    <T> T get(Object key, Class<T> type);
    <T> T get(Object key, Callable<T> valueLoader);
    void put(Object key, Object value);
    ValueWrapper putIfAbsent(Object key, Object value);
    void evict(Object key);
    void clear();
}

主要实现类包括：

ConcurrentMapCache：基于ConcurrentHashMap的简单实现

RedisCache：Redis集成实现

CaffeineCache：Caffeine缓存实现

1.2 RedisCache源码深入分析

RedisCache是Spring Data Redis提供的缓存实现，核心类结构如下：


public class RedisCache extends AbstractValueAdaptingCache {
    private final String name;
    private final RedisCacheWriter cacheWriter;
    private final RedisCacheConfiguration cacheConfig;

    // 构造函数和核心方法实现
}

核心组件解析：

RedisCacheWriter：负责与Redis通信的底层接口

RedisCacheConfiguration：缓存配置，如序列化器、TTL等

AbstractValueAdaptingCache：提供缓存值处理的基础实现

1.3 RedisCacheWriter源码分析

RedisCacheWriter是Redis操作的核心接口：


public interface RedisCacheWriter {
    void put(String name, byte[] key, byte[] value, @Nullable Duration ttl);
    byte[] get(String name, byte[] key);
    byte[] putIfAbsent(String name, byte[] key, byte[] value, @Nullable Duration ttl);
    void remove(String name, byte[] key);
    void clean(String name, byte[] pattern);
}

主要实现类：

DefaultRedisCacheWriter：基于RedisTemplate的默认实现

LettuceRedisCacheWriter：基于Lettuce客户端的优化实现

JedisRedisCacheWriter：基于Jedis客户端的实现

二、BatchCache接口设计与实现思路

2.1 BatchCache接口定义

为了实现批量操作能力，我们需要定义一个扩展接口：


/**
 * 扩展Spring Cache接口，提供批量操作能力
 *
 * @author doubao
 */
public interface BatchCache extends Cache {

    /**
     * 批量获取缓存值
     *
     * @param keys 缓存键集合
     * @return 键值对映射，不存在的键对应的值为null
     */
    Map<Object, Object> getAll(Collection<?> keys);

    /**
     * 批量存入缓存值
     *
     * @param values 键值对映射
     */
    void putAll(Map<?, ?> values);

    /**
     * 批量存入缓存值，仅当键不存在时
     *
     * @param values 键值对映射
     * @return 实际存入的键值对映射，已存在的键对应的值为null
     */
    Map<Object, Object> putAllIfAbsent(Map<?, ?> values);

    /**
     * 批量删除缓存
     *
     * @param keys 缓存键集合
     */
    void evictAll(Collection<?> keys);
}

2.2 实现思路概述

实现BatchCache接口的核心思路：

继承RedisCache类，复用现有功能

扩展RedisCacheWriter接口，添加批量操作方法

实现BatchRedisCache类，实现BatchCache接口

提供配置类，注册自定义CacheManager

确保与Spring Cache生态系统兼容

三、核心代码实现

3.1 扩展RedisCacheWriter接口


/**
 * 扩展RedisCacheWriter接口，添加批量操作方法
 *
 * @author doubao
 */
public interface BatchRedisCacheWriter extends RedisCacheWriter {

    /**
     * 批量获取缓存值
     *
     * @param name 缓存名称
     * @param keys 缓存键集合（字节数组形式）
     * @return 键值对映射，不存在的键对应的值为null
     */
    Map<byte[], byte[]> getAll(String name, Collection<byte[]> keys);

    /**
     * 批量存入缓存值
     *
     * @param name 缓存名称
     * @param values 键值对映射（字节数组形式）
     * @param ttl 过期时间，null表示不过期
     */
    void putAll(String name, Map<byte[], byte[]> values, @Nullable Duration ttl);

    /**
     * 批量删除缓存
     *
     * @param name 缓存名称
     * @param keys 缓存键集合（字节数组形式）
     */
    void removeAll(String name, Collection<byte[]> keys);
}

3.2 实现BatchRedisCacheWriter


/**
 * RedisCacheWriter的批量操作实现
 *
 * @author doubao
 */
public class DefaultBatchRedisCacheWriter implements BatchRedisCacheWriter {

    private final RedisTemplate<byte[], byte[]> redisTemplate;
    private final Duration sleepTime;

    /**
     * 构造函数
     *
     * @param redisOperations Redis操作模板
     */
    public DefaultBatchRedisCacheWriter(RedisOperations<byte[], byte[]> redisOperations) {
        this(redisOperations, Duration.ZERO);
    }

    /**
     * 构造函数
     *
     * @param redisOperations Redis操作模板
     * @param sleepTime 重试间隔时间
     */
    public DefaultBatchRedisCacheWriter(RedisOperations<byte[], byte[]> redisOperations, Duration sleepTime) {
        Assert.notNull(redisOperations, "RedisOperations must not be null!");
        Assert.notNull(sleepTime, "SleepTime must not be null!");

        this.redisTemplate = (RedisTemplate<byte[], byte[]>) redisOperations;
        this.sleepTime = sleepTime;
    }

    @Override
    public void put(String name, byte[] key, byte[] value, @Nullable Duration ttl) {
        execute(name, connection -> {
            if (shouldExpireWithin(ttl)) {
                connection.setEx(key, ttl.getSeconds(), value);
            } else {
                connection.set(key, value);
            }
            return "OK";
        });
    }

    @Override
    public byte[] get(String name, byte[] key) {
        return execute(name, connection -> connection.get(key));
    }

    // 其他方法实现...

    @Override
    public Map<byte[], byte[]> getAll(String name, Collection<byte[]> keys) {
        return execute(name, connection -> {
            List<byte[]> values = connection.mGet(keys.toArray(new byte[0][]));
            Map<byte[], byte[]> result = new LinkedHashMap<>(keys.size());

            int index = 0;
            for (byte[] key : keys) {
                result.put(key, index < values.size() ? values.get(index) : null);
                index++;
            }

            return result;
        });
    }

    @Override
    public void putAll(String name, Map<byte[], byte[]> values, @Nullable Duration ttl) {
        execute(name, connection -> {
            if (shouldExpireWithin(ttl)) {
                Pipeline pipeline = connection.openPipeline();
                for (Map.Entry<byte[], byte[]> entry : values.entrySet()) {
                    pipeline.setEx(entry.getKey(), ttl.getSeconds(), entry.getValue());
                }
                pipeline.close();
            } else {
                Map<byte[], byte[]> nonNullValues = values.entrySet().stream()
                    .filter(e -> e.getValue() != null)
                    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

                if (!nonNullValues.isEmpty()) {
                    connection.mSet(nonNullValues);
                }
            }
            return "OK";
        });
    }

    @Override
    public void removeAll(String name, Collection<byte[]> keys) {
        execute(name, connection -> {
            if (!keys.isEmpty()) {
                connection.del(keys.toArray(new byte[0][]));
            }
            return "OK";
        });
    }

    // 辅助方法...

    private <T> T execute(String name, RedisCallback<T> callback) {
        try {
            return redisTemplate.execute(callback);
        } catch (Exception ex) {
            throw new CacheOperationFailedException(name, "Redis batch operation failed", ex);
        }
    }

    private boolean shouldExpireWithin(@Nullable Duration ttl) {
        return ttl != null && !ttl.isZero() && !ttl.isNegative();
    }
}

3.3 实现BatchRedisCache类


/**
 * 支持批量操作的Redis缓存实现
 *
 * @author doubao
 */
public class BatchRedisCache extends RedisCache implements BatchCache {

    private final BatchRedisCacheWriter cacheWriter;
    private final RedisSerializationContext<Object, Object> serializationContext;

    /**
     * 构造函数
     *
     * @param name 缓存名称
     * @param cacheWriter 缓存写入器
     * @param cacheConfig 缓存配置
     */
    protected BatchRedisCache(String name, BatchRedisCacheWriter cacheWriter,
                             RedisCacheConfiguration cacheConfig) {
        super(name, cacheWriter, cacheConfig);
        this.cacheWriter = cacheWriter;
        this.serializationContext = cacheConfig.getSerializationContext();
    }

    @Override
    public Map<Object, Object> getAll(Collection<?> keys) {
        // 转换键为字节数组
        Map<Object, byte[]> keyMappings = new LinkedHashMap<>(keys.size());
        for (Object key : keys) {
            keyMappings.put(key, serializeCacheKey(createCacheKey(key)));
        }

        // 批量获取缓存值
        Map<byte[], byte[]> results = cacheWriter.getAll(
            getName(), keyMappings.values());

        // 反序列化结果
        Map<Object, Object> valueMappings = new LinkedHashMap<>(results.size());
        for (Map.Entry<Object, byte[]> entry : keyMappings.entrySet()) {
            byte[] valueBytes = results.get(entry.getValue());
            valueMappings.put(entry.getKey(), deserializeCacheValue(valueBytes));
        }

        return valueMappings;
    }

    @Override
    public void putAll(Map<?, ?> values) {
        // 序列化键值对
        Map<byte[], byte[]> serializedValues = new LinkedHashMap<>(values.size());
        for (Map.Entry<?, ?> entry : values.entrySet()) {
            if (entry.getValue() != null) {
                String cacheKey = createCacheKey(entry.getKey());
                byte[] keyBytes = serializeCacheKey(cacheKey);
                byte[] valueBytes = serializeCacheValue(entry.getValue());
                serializedValues.put(keyBytes, valueBytes);
            }
        }

        // 批量存入缓存
        cacheWriter.putAll(getName(), serializedValues, getTtl());
    }

    @Override
    public Map<Object, Object> putAllIfAbsent(Map<?, ?> values) {
        // 实现略，需要使用Redis事务或Lua脚本确保原子性
        throw new UnsupportedOperationException("Batch putIfAbsent operation is not supported yet.");
    }

    @Override
    public void evictAll(Collection<?> keys) {
        // 转换键为字节数组
        Collection<byte[]> keyBytes = keys.stream()
            .map(key -> serializeCacheKey(createCacheKey(key)))
            .collect(Collectors.toList());

        // 批量删除缓存
        cacheWriter.removeAll(getName(), keyBytes);
    }

    // 辅助方法...

    private byte[] serializeCacheKey(String cacheKey) {
        return serializationContext.getKeySerializationPair().write(cacheKey);
    }

    private byte[] serializeCacheValue(Object value) {
        return serializationContext.getValueSerializationPair().write(value);
    }

    private Object deserializeCacheValue(byte[] valueBytes) {
        if (valueBytes == null) {
            return null;
        }
        return serializationContext.getValueSerializationPair().read(valueBytes);
    }

    private String createCacheKey(Object key) {
        String convertedKey = convertKey(key);
        if (!getCacheConfiguration().usePrefix()) {
            return convertedKey;
        }
        return getCacheConfiguration().getKeyPrefixFor(name) + convertedKey;
    }
}

3.4 实现BatchRedisCacheManager


/**
 * 支持批量操作的Redis缓存管理器
 *
 * @author doubao
 */
public class BatchRedisCacheManager extends RedisCacheManager {

    private final BatchRedisCacheWriter cacheWriter;
    private final RedisCacheConfiguration defaultCacheConfig;

    /**
     * 构造函数
     *
     * @param cacheWriter 缓存写入器
     * @param defaultCacheConfiguration 默认缓存配置
     */
    public BatchRedisCacheManager(BatchRedisCacheWriter cacheWriter,
                                 RedisCacheConfiguration defaultCacheConfiguration) {
        super(cacheWriter, defaultCacheConfiguration);
        this.cacheWriter = cacheWriter;
        this.defaultCacheConfig = defaultCacheConfiguration;
    }

    @Override
    protected RedisCache createRedisCache(String name, @Nullable RedisCacheConfiguration cacheConfig) {
        return new BatchRedisCache(name, cacheWriter,
            cacheConfig != null ? cacheConfig : defaultCacheConfig);
    }

    /**
     * 从RedisCacheManager转换为BatchRedisCacheManager
     *
     * @param cacheManager Redis缓存管理器
     * @return 支持批量操作的Redis缓存管理器
     */
    public static BatchRedisCacheManager fromRedisCacheManager(RedisCacheManager cacheManager) {
        // 获取RedisCacheManager的私有字段
        Field cacheWriterField;
        Field defaultCacheConfigField;

        try {
            cacheWriterField = RedisCacheManager.class.getDeclaredField("cacheWriter");
            defaultCacheConfigField = RedisCacheManager.class.getDeclaredField("defaultCacheConfig");

            cacheWriterField.setAccessible(true);
            defaultCacheConfigField.setAccessible(true);

            RedisCacheWriter cacheWriter = (RedisCacheWriter) cacheWriterField.get(cacheManager);
            RedisCacheConfiguration defaultCacheConfig =
                (RedisCacheConfiguration) defaultCacheConfigField.get(cacheManager);

            // 创建BatchRedisCacheWriter
            BatchRedisCacheWriter batchCacheWriter;
            if (cacheWriter instanceof DefaultRedisCacheWriter) {
                DefaultRedisCacheWriter defaultWriter = (DefaultRedisCacheWriter) cacheWriter;
                // 使用反射获取RedisOperations
                Field redisOperationsField = DefaultRedisCacheWriter.class.getDeclaredField("redisOperations");
                redisOperationsField.setAccessible(true);
                RedisOperations<byte[], byte[]> redisOperations =
                    (RedisOperations<byte[], byte[]>) redisOperationsField.get(defaultWriter);

                batchCacheWriter = new DefaultBatchRedisCacheWriter(redisOperations);
            } else {
                // 回退方案，使用RedisTemplate创建
                RedisTemplate<byte[], byte[]> redisTemplate = new RedisTemplate<>();
                redisTemplate.setConnectionFactory(cacheManager.getCacheWriter().getConnectionFactory());
                redisTemplate.afterPropertiesSet();
                batchCacheWriter = new DefaultBatchRedisCacheWriter(redisTemplate);
            }

            // 创建BatchRedisCacheManager
            return new BatchRedisCacheManager(batchCacheWriter, defaultCacheConfig);

        } catch (NoSuchFieldException | IllegalAccessException e) {
            throw new RuntimeException("Failed to convert RedisCacheManager to BatchRedisCacheManager", e);
        }
    }
}

3.5 配置类实现


/**
 * 批量缓存配置类
 *
 * @author doubao
 */
@Configuration
public class BatchCacheConfiguration {

    @Bean
    public BatchRedisCacheManager batchRedisCacheManager(RedisConnectionFactory redisConnectionFactory) {
        // 创建默认配置
        RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig()
            .entryTtl(Duration.ofMinutes(10))
            .serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(new StringRedisSerializer()))
            .serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(new GenericJackson2JsonRedisSerializer()));

        // 创建BatchRedisCacheWriter
        RedisTemplate<byte[], byte[]> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisConnectionFactory);
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());
        redisTemplate.afterPropertiesSet();

        BatchRedisCacheWriter cacheWriter = new DefaultBatchRedisCacheWriter(redisTemplate);

        // 创建BatchRedisCacheManager
        BatchRedisCacheManager cacheManager = new BatchRedisCacheManager(cacheWriter, config);
        cacheManager.setTransactionAware(true);

        // 设置缓存名称和配置的映射
        Map<String, RedisCacheConfiguration> cacheConfigurations = new HashMap<>();
        // 可以为不同的缓存名称配置不同的策略
        cacheConfigurations.put("batchCache", config.entryTtl(Duration.ofHours(1)));

        cacheManager.setCacheConfigurations(cacheConfigurations);

        return cacheManager;
    }

    /**
     * 自定义CacheResolver，支持BatchCache
     *
     * @param cacheManager 缓存管理器
     * @return 缓存解析器
     */
    @Bean
    public CacheResolver batchCacheResolver(BatchRedisCacheManager cacheManager) {
        return new SimpleCacheResolver(cacheManager) {
            @Override
            protected Collection<? extends Cache> getCaches(CacheOperationInvocationContext<?> context) {
                Collection<? extends Cache> caches = super.getCaches(context);
                return caches.stream()
                    .map(cache -> {
                        if (cache instanceof RedisCache && !(cache instanceof BatchRedisCache)) {
                            // 将普通RedisCache转换为BatchRedisCache
                            RedisCache redisCache = (RedisCache) cache;
                            return new BatchRedisCache(
                                redisCache.getName(),
                                (BatchRedisCacheWriter) cacheManager.getCacheWriter(),
                                redisCache.getCacheConfiguration());
                        }
                        return cache;
                    })
                    .collect(Collectors.toList());
            }
        };
    }
}

四、使用示例

4.1 定义业务服务


/**
 * 示例服务类，演示BatchCache的使用
 *
 * @author doubao
 */
@Service
public class UserService {

    private final UserRepository userRepository;

    @Autowired
    public UserService(UserRepository userRepository) {
        this.userRepository = userRepository;
    }

    /**
     * 批量获取用户信息，使用缓存
     *
     * @param userIds 用户ID集合
     * @return 用户信息映射
     */
    @Cacheable(value = "users", key = "#root.methodName + '_' + #userIds", unless = "#result == null")
    public Map<Long, User> getUsersBatch(Collection<Long> userIds) {
        // 模拟从数据库获取数据
        return userIds.stream()
            .collect(Collectors.toMap(
                Function.identity(),
                userId -> userRepository.findById(userId).orElse(null)
            ));
    }

    /**
     * 批量保存用户信息，并更新缓存
     *
     * @param users 用户信息集合
     */
    @CachePut(value = "users", key = "#root.methodName + '_' + #users.![id]", condition = "#users != null && !#users.isEmpty()")
    public Map<Long, User> saveUsersBatch(Collection<User> users) {
        // 模拟批量保存到数据库
        users.forEach(userRepository::save);

        // 返回保存后的用户信息
        return users.stream()
            .collect(Collectors.toMap(User::getId, Function.identity()));
    }

    /**
     * 批量删除用户信息，并清除缓存
     *
     * @param userIds 用户ID集合
     */
    @CacheEvict(value = "users", allEntries = false, key = "#root.methodName + '_' + #userIds")
    public void deleteUsersBatch(Collection<Long> userIds) {
        // 模拟批量删除
        userIds.forEach(userRepository::deleteById);
    }

    /**
     * 直接使用BatchCache接口的批量操作
     *
     * @param userIds 用户ID集合
     * @return 用户信息映射
     */
    public Map<Long, User> getUsersBatchWithBatchCache(Collection<Long> userIds) {
        // 通过ApplicationContext获取BatchCache
        BatchCache batchCache = (BatchCache) applicationContext.getBean("cacheManager").getCache("users");

        // 直接使用BatchCache的批量获取方法
        Map<Object, Object> cacheResults = batchCache.getAll(userIds);

        // 处理缓存结果
        Map<Long, User> result = new HashMap<>();
        for (Map.Entry<Object, Object> entry : cacheResults.entrySet()) {
            Long userId = (Long) entry.getKey();
            User user = (User) entry.getValue();

            if (user == null) {
                // 缓存未命中，从数据库获取
                user = userRepository.findById(userId).orElse(null);
                if (user != null) {
                    // 手动放入缓存
                    batchCache.put(userId, user);
                }
            }

            result.put(userId, user);
        }

        return result;
    }
}

4.2 配置文件示例


spring:
  redis:
    host: localhost
    port: 6379
    password:
    timeout: 10000ms
    lettuce:
      pool:
        max-active: 8
        max-wait: -1ms
        max-idle: 8
        min-idle: 0
  cache:
    type: redis
    redis:
      time-to-live: 600000  # 10分钟
      cache-null-values: true
      use-key-prefix: true
      key-prefix: batch_cache:

五、性能测试与优化

5.1 性能测试框架


/**
 * BatchCache性能测试
 *
 * @author doubao
 */
@SpringBootTest
public class BatchCachePerformanceTest {

    @Autowired
    private UserService userService;

    @Autowired
    private CacheManager cacheManager;

    private static final int TEST_SIZE = 1000;
    private static final int WARMUP_TIMES = 10;
    private static final int TEST_TIMES = 100;

    @BeforeEach
    public void setUp() {
        // 准备测试数据
        List<User> users = new ArrayList<>(TEST_SIZE);
        for (int i = 0; i < TEST_SIZE; i++) {
            User user = new User();
            user.setId((long) i);
            user.setName("User" + i);
            user.setAge(20 + i % 30);
            users.add(user);
        }

        // 预热
        for (int i = 0; i < WARMUP_TIMES; i++) {
            userService.saveUsersBatch(users);
            userService.getUsersBatch(users.stream().map(User::getId).collect(Collectors.toList()));
        }

        // 清除缓存
        Cache usersCache = cacheManager.getCache("users");
        if (usersCache != null) {
            usersCache.clear();
        }
    }

    @Test
    public void testBatchGetPerformance() {
        List<Long> userIds = IntStream.range(0, TEST_SIZE)
            .mapToObj(Long::valueOf)
            .collect(Collectors.toList());

        // 测试单键获取性能
        long singleStartTime = System.currentTimeMillis();
        for (int i = 0; i < TEST_TIMES; i++) {
            for (Long userId : userIds) {
                userService.getUser(userId);
            }
        }
        long singleEndTime = System.currentTimeMillis();
        long singleTotalTime = singleEndTime - singleStartTime;

        // 测试批量获取性能
        long batchStartTime = System.currentTimeMillis();
        for (int i = 0; i < TEST_TIMES; i++) {
            userService.getUsersBatch(userIds);
        }
        long batchEndTime = System.currentTimeMillis();
        long batchTotalTime = batchEndTime - batchStartTime;

        // 输出性能结果
        System.out.printf("单键获取 %d 次，总耗时: %d ms，平均每次: %f ms%n",
            TEST_SIZE * TEST_TIMES, singleTotalTime, (double) singleTotalTime / (TEST_SIZE * TEST_TIMES));
        System.out.printf("批量获取 %d 次，总耗时: %d ms，平均每次: %f ms%n",
            TEST_TIMES, batchTotalTime, (double) batchTotalTime / TEST_TIMES);
        System.out.printf("批量操作性能提升: %.2f%%%n",
            (1 - (double) batchTotalTime / singleTotalTime) * 100);
    }

    // 其他性能测试方法...
}

5.2 性能优化策略

批量操作优化：

使用Redis的MGET、MSET等批量命令

合理设置批量操作的大小，避免单次操作过大

考虑使用Redis Pipeline提升性能

序列化优化：

使用高效的序列化方式，如Kryo、Protostuff等

避免序列化大对象，可考虑拆分数据

缓存配置优化：

根据业务场景设置合理的TTL

使用分区缓存，避免不同业务数据相互影响

考虑使用二级缓存（如Caffeine + Redis）提升性能

六、注意事项与最佳实践

6.1 使用注意事项

事务支持：

Spring Cache的@Cacheable、@CachePut等注解不支持事务回滚

如果需要事务支持，建议在业务代码中直接使用BatchCache接口

异常处理：

批量操作可能部分成功部分失败，需要业务层处理这种情况

考虑实现重试机制，确保操作的最终一致性

缓存穿透与雪崩：

批量操作同样需要防范缓存穿透问题

合理设置不同数据的TTL，避免缓存雪崩

6.2 最佳实践

批量操作大小控制：

对于大量数据的批量操作，建议分批处理

每批大小可根据网络情况和Redis性能调整，一般建议在100-1000之间

缓存预热：

对于热点数据，启动时进行缓存预热

使用BatchCache的批量操作能力快速填充缓存

监控与告警：

监控批量操作的性能指标，如QPS、响应时间等

设置合理的告警阈值，及时发现性能问题

通过以上方案，我们成功地扩展了Spring Boot Cache的功能，实现了兼容RedisCache的BatchCache接口。这种实现方式不仅保持了与Spring Cache生态的兼容性，还显著提升了批量数据操作的性能，为高并发场景下的应用提供了有力支持。