Spring Boot集群中如何确保Hazelcast EntryProcessor在调用节点执行？

Hazelcast的EntryProcessor默认是在数据的主节点执行的，这是分布式缓存的设计初衷——就近操作数据避免网络开销。如果要强制在发起调用的节点执行，需要绕开默认的分区路由逻辑，以下是几种可行方案：

1. 自定义分区策略，绑定Key到发起节点的分区 #

通过自定义PartitionStrategy，让指定Key强制映射到发起调用节点负责的分区，从而让该节点成为Key的主节点，EntryProcessor自然会在本地执行。

实现步骤： #

• 自定义分区策略类，识别绑定节点的Key并路由到指定分区：

import com.hazelcast.core.PartitionStrategy;
import java.util.Objects;

public class NodeBoundPartitionStrategy implements PartitionStrategy {
    @Override
    public int getPartition(Object key) {
        if (key instanceof NodeBoundKey) {
            return ((NodeBoundKey) key).getTargetPartitionId();
        }
        // 非绑定Key沿用默认哈希路由
        return key.hashCode();
    }

    // 封装绑定目标分区的Key
    public static class NodeBoundKey {
        private final Object actualKey;
        private final int targetPartitionId;

        public NodeBoundKey(Object actualKey, int targetPartitionId) {
            this.actualKey = actualKey;
            this.targetPartitionId = targetPartitionId;
        }

        public int getTargetPartitionId() {
            return targetPartitionId;
        }

        // 保证实际Key的唯一性（重写equals和hashCode）
        @Override
        public boolean equals(Object o) {
            if (this == o) return true;
            if (o == null || getClass() != o.getClass()) return false;
            NodeBoundKey that = (NodeBoundKey) o;
            return Objects.equals(actualKey, that.actualKey);
        }

        @Override
        public int hashCode() {
            return Objects.hash(actualKey);
        }
    }
}

• 在Hazelcast配置中指定该策略：

com.hazelcast.config.Config hazelcastConfig = new com.hazelcast.config.Config();
hazelcastConfig.getMapConfig("your-cache-map")
               .setPartitionStrategy(new NodeBoundPartitionStrategy());

• 调用时，将实际Key包装为绑定本地分区的NodeBoundKey：

import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.Member;
import com.hazelcast.core.Partition;
import com.hazelcast.core.PartitionService;

// 获取本地节点及分区服务
HazelcastInstance hazelcastInstance = ...; // 注入或获取实例
PartitionService partitionService = hazelcastInstance.getPartitionService();
Member localMember = hazelcastInstance.getCluster().getLocalMember();

// 找到本地节点负责的任意一个分区ID
int localPartitionId = partitionService.getPartitions().stream()
        .filter(partition -> partition.getOwner().equals(localMember))
        .findFirst()
        .map(Partition::getPartitionId)
        .orElseThrow(() -> new IllegalStateException("本地节点未负责任何分区"));

// 包装Key并执行EntryProcessor
NodeBoundPartitionStrategy.NodeBoundKey boundKey = 
        new NodeBoundPartitionStrategy.NodeBoundKey("your-actual-key", localPartitionId);
yourCacheMap.executeOnKey(boundKey, yourEntryProcessor);

注意：这种方式会打破Hazelcast的负载均衡机制，可能导致大量Key集中在少数节点，引发数据倾斜和性能瓶颈，仅适合必须本地执行的特定场景。

2. 本地缓存+分布式事件同步 #

放弃分布式EntryProcessor，改用本地缓存处理更新，再通过Hazelcast的消息机制同步到集群其他节点。这种方式让更新逻辑完全在发起节点执行，同时保证集群缓存的最终一致性。

实现示例： #

import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.ITopic;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.LoadingCache;

import java.util.concurrent.TimeUnit;

public class LocalCacheWithSync {
    // 本地缓存（用Guava Cache举例）
    private final LoadingCache<String, Object> localCache;
    // Hazelcast消息主题，用于同步更新事件
    private final ITopic<CacheUpdateEvent> updateTopic;
    private final HazelcastInstance hazelcastInstance;
    private final String distributedMapName;

    public LocalCacheWithSync(HazelcastInstance hazelcastInstance, String distributedMapName) {
        this.hazelcastInstance = hazelcastInstance;
        this.distributedMapName = distributedMapName;
        // 初始化本地缓存，默认从分布式Map加载数据
        this.localCache = CacheBuilder.newBuilder()
                .expireAfterWrite(10, TimeUnit.MINUTES)
                .build(key -> hazelcastInstance.getMap(distributedMapName).get(key));
        // 初始化更新主题并订阅同步事件
        this.updateTopic = hazelcastInstance.getTopic("cache-update-topic-" + distributedMapName);
        this.subscribeToUpdateEvents();
    }

    // 本地更新缓存并发布同步事件
    public void updateCache(String key, Object newValue) {
        localCache.put(key, newValue);
        // 更新分布式Map保证初始加载正确性
        hazelcastInstance.getMap(distributedMapName).put(key, newValue);
        // 发布更新事件到集群
        updateTopic.publish(new CacheUpdateEvent(key, newValue));
    }

    // 订阅集群更新事件，同步本地缓存
    private void subscribeToUpdateEvents() {
        updateTopic.addMessageListener(message -> {
            CacheUpdateEvent event = message.getMessageObject();
            localCache.put(event.getKey(), event.getValue());
        });
    }

    // 缓存更新事件（需实现Serializable）
    public static class CacheUpdateEvent implements java.io.Serializable {
        private String key;
        private Object value;

        public CacheUpdateEvent(String key, Object value) {
            this.key = key;
            this.value = value;
        }

        // getter和setter
        public String getKey() { return key; }
        public Object getValue() { return value; }
    }
}

适用场景：更新频繁、对一致性要求可放宽为最终一致，且能接受本地缓存与分布式Map短暂不一致的业务。

3. 注意事项 #

Hazelcast设计EntryProcessor在主节点执行是为了最大化性能，强制本地执行本质上是反其设计初衷的操作。在选择方案前，请务必评估：

• 数据倾斜风险（方案1）
• 一致性保障成本（方案2）
• 集群负载均衡的影响

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