我正在尝试使用 infinispan 缓存实现计数器系统。多个线程试图访问同一个键并针对该键增加值。
计数器操作以 lock(key) 开始,因此没有其他线程可以写入它。获得锁的线程执行 get(key),递增值并将其放回同一个键。
另一个等待锁释放的线程调用 get(key)。get(KEY) 返回先前的值,而不是预期的更新计数器值。
我正在针对每个线程中的缓存开始一个新事务。事务是自动提交的。
我尝试使用自动提交以及在放置后显式提交。我还确保在每次运行之前清除文件持久性和数据库持久性。
这是我的 infinispan 缓存配置。
<distributed-cache name="COUNTER_CACHE">
<transaction transaction-manager-lookup="com.suntecgroup.tbms.container.services.cluster.ContainerCacheTxManagerLookup" mode="FULL_XA" locking="PESSIMISTIC"/>
<state-transfer timeout="360000" await-initial-transfer="true" enabled="true"/>
<locking isolation="READ_COMMITTED" acquire-timeout="60000" concurrency-level="100" striping="false"/>
<memory>
<off-heap size="1000" eviction="COUNT"/>
<!-- Determines the max entries to hold in the cache -->
</memory>
<persistence passivation="false">
<file-store path="./infinispan_persistance_tmp/" max-entries="1000"/>
<store class="com.suntecgroup.tbms.container.cache.store.ispn.InfinispanCounterCacheStore" fetch-state="false" preload="true" shared="false" purge="false" read-only="false" singleton="false"> </store>
</persistence>
</distributed-cache>
预期的结果是,当另一个线程尝试获取相同的值时,应该反映一个线程对值所做的任何更新。
我最好的猜测是它必须对每个线程创建的事务做一些事情。
我无法分享实际代码,但以下代码近似于该过程。希望这可以帮助
class CounterThread implements Runnable{
public Integer getCounterVal(String key){
int counter = cacheDelegate.get(key);
if(counter==null){
synchronized (Counter.class) {
if(counter = cacheDelegate.beginTransactionIfNoneAndGetObjectWithLock(key)){
counter = 0;
cacheDelegate.putAndCommit(key,counter);
}
}
} else
counter = cacheDelegate.beginTransactionIfNoneAndGetObjectWithLock(key);
return counter;
}
public void putCounterVal(String key,int val){
val++;
cacheDelegate.putAndCommit(key,val);
}
public void run(){
int i = 0;
while(i<100){
int counter = getCounterVal("KEY");
putCounterVal("KEY",counter);
i++;
}
}
}
// this method is used in beginTransactionIfNoneAndGetObjectWithLock to create a transaction
public void begin()
throws ContainerPlatformServicesException {
try {
if (getCache("COUNTER_CACHE").getAdvancedCache().getTransactionManager()
.getTransaction() == null) {
getCache("COUNTER_CACHE").getAdvancedCache().getTransactionManager()
.setTransactionTimeout(transactionTimeOut);
getCache("COUNTER_CACHE").getAdvancedCache().getTransactionManager()
.begin();
}
}catch(TimeoutException e){
logger.warn("TimeoutException exception caught. Required node is suspected. Thus waiting for suspect completion, to begin transaction.");
}
catch (Exception e) {
logger.error("Failed to begin transaction. Details:-", e);
}
}
// Only two threads are running.