diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java index bb157c55fa..7e0e655f6d 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java @@ -18,12 +18,16 @@ import java.sql.SQLException; import java.sql.Types; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReadWriteLock; +import java.util.concurrent.locks.ReentrantReadWriteLock; import javax.sql.DataSource; @@ -89,13 +93,21 @@ import org.springframework.util.StringUtils; *

* @author Gunnar Hillert + * @author Artem Bilan * @since 2.2 */ @ManagedResource public class JdbcChannelMessageStore extends AbstractMessageGroupStore implements InitializingBean { private static final Log logger = LogFactory.getLog(JdbcChannelMessageStore.class); - private final Set idCache = Collections.newSetFromMap(new ConcurrentHashMap()); + + private final Set idCache = new HashSet(); + + private final ReadWriteLock idCacheLock = new ReentrantReadWriteLock(); + + private final Lock idCacheReadLock = idCacheLock.readLock(); + + private final Lock idCacheWriteLock = idCacheLock.writeLock(); /** * Default value for the table prefix property. @@ -444,16 +456,22 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore implement final String query; - synchronized (idCache) { + final List> messages; + + this.idCacheReadLock.lock(); + try { if (this.usingIdCache && !this.idCache.isEmpty()) { query = getQuery(this.channelMessageStoreQueryProvider.getPollFromGroupExcludeIdsQuery()); parameters.addValue("message_ids", idCache); } else { query = getQuery(this.channelMessageStoreQueryProvider.getPollFromGroupQuery()); } + messages = namedParameterJdbcTemplate.query(query, parameters, messageRowMapper); + } + finally { + this.idCacheReadLock.unlock(); } - final List> messages = namedParameterJdbcTemplate.query(query, parameters, messageRowMapper); Assert.isTrue(messages.size() == 0 || messages.size() == 1); if (messages.size() > 0){ @@ -462,11 +480,16 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore implement final String messageId = message.getHeaders().getId().toString(); if (this.usingIdCache) { + this.idCacheWriteLock.lock(); + try { + boolean added = this.idCache.add(messageId); - boolean added = this.idCache.add(messageId); - - if (logger.isDebugEnabled()) { - logger.debug(String.format("Polled message with id '%s' added: '%s'.", messageId, added)); + if (logger.isDebugEnabled()) { + logger.debug(String.format("Polled message with id '%s' added: '%s'.", messageId, added)); + } + } + finally { + this.idCacheWriteLock.unlock(); } } @@ -612,7 +635,13 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore implement if (logger.isDebugEnabled()) { logger.debug("Removing Message Id:" + messageId); } - this.idCache.remove(messageId); + this.idCacheWriteLock.lock(); + try { + this.idCache.remove(messageId); + } + finally { + this.idCacheWriteLock.unlock(); + } } /** diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractTxTimeoutMessageStoreTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractTxTimeoutMessageStoreTests.java index d9afa16144..712b5bff52 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractTxTimeoutMessageStoreTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractTxTimeoutMessageStoreTests.java @@ -12,6 +12,15 @@ */ package org.springframework.integration.jdbc.store.channel; +import static org.junit.Assert.assertTrue; + +import java.util.concurrent.Callable; +import java.util.concurrent.CompletionService; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorCompletionService; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import javax.sql.DataSource; import org.junit.Assert; @@ -19,20 +28,22 @@ import org.junit.Assert; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; import org.springframework.integration.jdbc.store.JdbcChannelMessageStore; +import org.springframework.integration.message.GenericMessage; import org.springframework.integration.support.MessageBuilder; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.TransactionDefinition; import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.annotation.Isolation; +import org.springframework.transaction.support.TransactionCallback; import org.springframework.transaction.support.TransactionCallbackWithoutResult; import org.springframework.transaction.support.TransactionTemplate; /** - * * @author Gunnar Hillert - * + * @author Artem Bilan */ abstract class AbstractTxTimeoutMessageStoreTests { @@ -89,4 +100,61 @@ abstract class AbstractTxTimeoutMessageStoreTests { Assert.assertEquals(Integer.valueOf(0), Integer.valueOf(testService.getDuplicateMessagesCount())); } + public void testInt2993IdCacheConcurrency() throws InterruptedException, ExecutionException { + final String groupId = "testInt2993Group"; + for (int i = 0; i < 100; i++) { + this.jdbcChannelMessageStore.addMessageToGroup(groupId, new GenericMessage("testInt2993Message")); + } + + ExecutorService executorService = Executors.newCachedThreadPool(); + CompletionService completionService = new ExecutorCompletionService(executorService); + + final int concurrency = 5; + + final TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager); + + for (int i = 0; i < concurrency; i++) { + completionService.submit(new Callable() { + @Override + public Boolean call() throws Exception { + for (int i = 0; i < 100; i++) { + boolean result = transactionTemplate.execute(new TransactionCallback() { + @Override + public Boolean doInTransaction(TransactionStatus status) { + Message message = null; + try { + message = jdbcChannelMessageStore.pollMessageFromGroup(groupId); + } + catch (Exception e) { + log.error("IdCache race condition.", e); + return false; + } + try { + Thread.sleep(10); + } + catch (InterruptedException e) { + log.error(e); + } + if (message != null) { + jdbcChannelMessageStore.removeFromIdCache(message.getHeaders().getId().toString()); + } + return true; + } + }); + if (!result) return false; + } + + return true; + } + }); + } + + for (int j = 0; j < concurrency; j++) { + assertTrue(completionService.take().get()); + } + + executorService.shutdown(); + assertTrue(executorService.awaitTermination(5, TimeUnit.SECONDS)); + } + } diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/HsqlTxTimeoutMessageStoreTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/HsqlTxTimeoutMessageStoreTests.java index 8b2155470d..803a31af9f 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/HsqlTxTimeoutMessageStoreTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/HsqlTxTimeoutMessageStoreTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at @@ -12,6 +12,8 @@ */ package org.springframework.integration.jdbc.store.channel; +import java.util.concurrent.ExecutionException; + import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.test.context.ContextConfiguration; @@ -29,9 +31,13 @@ public class HsqlTxTimeoutMessageStoreTests extends AbstractTxTimeoutMessageStor @Test @Override public void test() throws InterruptedException { - super.test(); + } + @Test + @Override + public void testInt2993IdCacheConcurrency() throws InterruptedException, ExecutionException { + super.testInt2993IdCacheConcurrency(); } }