Hazelcast 4.2 反应式PostgreSQL WriteThrough实现及注入问题求助

核心问题原因 #

Hazelcast默认会自行初始化MapStore实例，该实例不属于Spring上下文管理范畴，因此直接用@Autowired注入Repository会出现null值。解决核心是让Spring接管MapStore的生命周期，确保Repository能被正确注入。

步骤1：依赖配置（pom.xml） #

<dependencies>
    <!-- Hazelcast核心与Spring集成 -->
    <dependency>
        <groupId>com.hazelcast</groupId>
        <artifactId>hazelcast</artifactId>
        <version>4.2</version>
    </dependency>
    <dependency>
        <groupId>com.hazelcast</groupId>
        <artifactId>hazelcast-spring</artifactId>
        <version>4.2</version>
    </dependency>

    <!-- Spring Reactive + PostgreSQL R2DBC驱动 -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-r2dbc</artifactId>
    </dependency>
    <dependency>
        <groupId>io.r2dbc</groupId>
        <artifactId>r2dbc-postgresql</artifactId>
    </dependency>
</dependencies>

步骤2：实体类与Reactive Repository #

// PostgreSQL对应实体类
@Table("user_info")
public class User {
    @Id
    private Long id;
    private String username;
    private String email;

    // 构造器、getter、setter省略
}

// 反应式Repository接口
public interface UserRepository extends ReactiveCrudRepository<User, Long> {
}

步骤3：Spring管理的MapStore实现 #

通过构造器注入Repository，确保Spring能正确传递实例：

@Component
public class UserMapStore implements MapStore<Long, User> {

    private final UserRepository userRepository;

    // 构造器注入，Spring自动注入Repository实例
    public UserMapStore(UserRepository userRepository) {
        this.userRepository = userRepository;
    }

    @Override
    public void delete(Long key) {
        // MapStore方法为同步阻塞，需调用block()等待反应式操作完成
        userRepository.deleteById(key).block();
    }

    @Override
    public void deleteAll(Collection<Long> keys) {
        userRepository.deleteAllById(keys).block();
    }

    @Override
    public void store(Long key, User value) {
        userRepository.save(value).block();
    }

    @Override
    public void storeAll(Map<Long, User> map) {
        userRepository.saveAll(map.values()).blockLast();
    }

    @Override
    public User load(Long key) {
        return userRepository.findById(key).block();
    }

    @Override
    public Map<Long, User> loadAll(Collection<Long> keys) {
        return userRepository.findAllById(keys)
                .collectMap(User::getId)
                .block();
    }

    @Override
    public Set<Long> loadAllKeys() {
        return userRepository.findAll()
                .map(User::getId)
                .collect(Collectors.toSet())
                .block();
    }
}

步骤4：Hazelcast配置类 #

将Spring管理的MapStore注入到Hazelcast缓存配置中：

@Configuration
public class HazelcastConfig {

    private final UserMapStore userMapStore;

    public HazelcastConfig(UserMapStore userMapStore) {
        this.userMapStore = userMapStore;
    }

    @Bean
    public HazelcastInstance hazelcastInstance() {
        Config config = new Config();
        MapConfig mapConfig = config.getMapConfig("user-cache");

        // 配置WriteThrough模式（WriteBehind只需调整writeDelaySeconds为大于0的值）
        MapStoreConfig mapStoreConfig = new MapStoreConfig()
                .setImplementation(userMapStore)
                .setWriteDelaySeconds(0) // 0=WriteThrough，>0=WriteBehind延迟秒数
                .setWriteBatchSize(10) // WriteBehind模式下的批量写入大小
                .setWriteCoalescing(true); // WriteBehind模式下合并相同key的重复操作

        mapConfig.setMapStoreConfig(mapStoreConfig);
        return Hazelcast.newHazelcastInstance(config);
    }
}

步骤5：PostgreSQL R2DBC配置（application.yml） #

spring:
  r2dbc:
    url: r2dbc:postgresql://localhost:5432/testdb
    username: postgres
    password: your-db-password

关键注意事项 #

• MapStore的方法为同步阻塞设计，因此反应式Repository的操作必须调用block()/blockLast()等待完成，否则Hazelcast无法感知操作结果。
• 必须给UserMapStore添加@Component注解，确保Spring接管其生命周期，才能完成Repository的注入。
• WriteBehind模式下，可通过调整writeDelaySeconds（如设置为5）和writeBatchSize来优化批量写入性能。

内容的提问来源于stack exchange，提问作者jeni A