Infinispan 分布式缓存中的并发更新：用户计数场景的同步策略

import org.infinispan.client.hotrod.RemoteCacheManager;
import org.infinispan.counter.api.CounterManager;
import org.infinispan.counter.api.StrongCounter;
import org.infinispan.counter.api.CounterConfiguration;
import org.infinispan.counter.api.CounterType;

public class UserCounterService {

    private final StrongCounter onlineUsersCounter;

    public UserCounterService(RemoteCacheManager cacheManager) {
        CounterManager counterManager = cacheManager.get
        CounterManager();

        // 尝试获取计数器，如果不存在则创建并配置
        // 确保计数器只被创建一次，例如在应用启动时
        if (!counterManager.is -->
        Defined("onlineUsers")) {
            counterManager.defineCounter("onlineUsers", 
                CounterConfiguration.builder(CounterType.UNBOUNDED_STRONG)
                    .initialValue(0)
                    .build());
        }
        this.onlineUsersCounter = counterManager.getStrongCounter("onlineUsers");
    }

    /**
     * 用户登录时调用，原子性地增加在线用户数
     */
    public long userLoggedIn() {
        return onlineUsersCounter.incrementAndGet().join(); // .join() 用于等待异步操作完成
    }

    /**
     * 用户登出时调用，原子性地减少在线用户数
     */
    public long userLoggedOut() {
        return onlineUsersCounter.decrementAndGet().join();
    }

    /**
     * 获取当前在线用户数
     */
    public long getCurrentOnlineUsers() {
        return onlineUsersCounter.getValue().join();
    }

    public static void main(String[] args) {
        // 假设已经配置并启动了 RemoteCacheManager
        // RemoteCacheManager cacheManager = new RemoteCacheManager("127.0.0.1:11222");
        // UserCounterService service = new UserCounterService(cacheManager);

        // 模拟用户登录
        // System.out.println("User logged in, current count: " + service.userLoggedIn());
        // System.out.println("User logged in, current count: " + service.userLoggedIn());
        // System.out.println("Current online users: " + service.getCurrentOnlineUsers());

        // cacheManager.stop();
    }
}

import org.infinispan.client.hotrod.RemoteCache;
import org.infinispan.client.hotrod.RemoteCacheManager;
import org.infinispan.client.hotrod.exceptions.HotRodClientException;
import org.infinispan.client.hotrod.transaction.manager.RemoteTransactionManager;
import jakarta.transaction.SystemException;
import jakarta.transaction.UserTransaction;

public class UserCountWithTransaction {

    private final RemoteCache<String, Long> userCountCache;
    private final RemoteTransactionManager transactionManager;

    public UserCountWithTransaction(RemoteCacheManager cacheManager) {
        this.userCountCache = cacheManager.getCache("userCountsCache");
        this.transactionManager = RemoteTransactionManager.getInstance(cacheManager);
    }

    public long incrementOnlineUsers() throws SystemException {
        UserTransaction userTransaction = transactionManager.getUserTransaction();
        try {
            userTransaction.begin(); // 开始事务

            Long currentCount = userCountCache.get("onlineUsers");
            if (currentCount == null) {
                currentCount = 0L;
            }
            Long newCount = currentCount + 1;
            userCountCache.put("onlineUsers", newCount); // 在事务内修改

            userTransaction.commit(); // 提交事务
            return newCount;
        } catch (Exception e) {
            if (userTransaction != null) {
                userTransaction.rollback(); // 发生异常时回滚事务
            }
            throw new RuntimeException("Failed to increment online users with transaction", e);
        }
    }

    public long getCurrentOnlineUsers() {
        return userCountCache.get("onlineUsers") == null ? 0L : userCountCache.get("onlineUsers");
    }
}

import org.infinispan.client.hotrod.RemoteCache;
import org.infinispan.client.hotrod.RemoteCacheManager;

public class UserCountWithVersionedOperations {

    private final RemoteCache<String, Long> userCountCache;

    public UserCountWithVersionedOperations(RemoteCacheManager cacheManager) {
        this.userCountCache = cacheManager.getCache("userCountsCache");
        // 确保缓存中有一个初始值
        userCountCache.putIfAbsent("onlineUsers", 0L);
    }

    public long incrementOnlineUsers() {
        while (true) { // 循环直到更新成功
            Long currentCount = userCountCache.get("onlineUsers");
            if (currentCount == null) {
                currentCount = 0L;
            }
            Long newCount = currentCount + 1;

            // 尝试原子性地替换旧值
            boolean success = userCountCache.replace("onlineUsers", currentCount, newCount);
            if (success) {
                return newCount;
            }
            // 如果替换失败，说明在读取 currentCount 后有其他线程修改了值，需要重试
            // System.out.println("Concurrent update detected, retrying increment...");
        }
    }

    public long getCurrentOnlineUsers() {
        return userCountCache.get("onlineUsers") == null ? 0L : userCountCache.get("onlineUsers");
    }
}