Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4048: Add JdbcLockRegistry support

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4048

The semantics of the locks are the same as for the default lock
registry (they have to be unlocked by the same thread that
locked them).

Updates from feedback

Add interface for JdbcClient so transactions can bind to proxy

Fix typo in exception message

Fix drop script

JdbcClient -> LockRepository

Make lock() contract more like native Lock

Make lockInterruptibly throw InterruptedException more

Delete all expired locks, not just the ones that we own

INT-4048: Polishing

* `DefaultLockRepository`: replace `prefix` only once in the `afterPropertiesSet()`
* Add JavaDocs to the `DefaultLockRepository`
* `JdbcLockRegistry`: add `Thread.sleep(100)` to lock loops do not request DB so often like in case of clean `while(true)`
* `JdbcLockRegistry`: implement more robust `tryLock(long time, TimeUnit unit)`. The logic mostly copied from `RedisLockRegistry`
* Add `testExclusiveAccess()` to be sure that JDBC locks have exclusive access to the data sequence

Document `JdbcLockRegistry`

Upgrade to SF-4.3 GA
This commit is contained in:
Dave Syer
2016-06-07 14:53:01 +01:00
committed by Artem Bilan
parent eca8536678
commit 7c2731dbb9
31 changed files with 1217 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -141,7 +141,7 @@ subprojects { subproject ->
springSecurityVersion = '4.1.0.RELEASE'
springSocialTwitterVersion = '1.1.2.RELEASE'
springRetryVersion = '1.1.2.RELEASE'
springVersion = project.hasProperty('springVersion') ? project.springVersion : '4.3.0.BUILD-SNAPSHOT'
springVersion = project.hasProperty('springVersion') ? project.springVersion : '4.3.0.RELEASE'
springWsVersion = '2.3.0.RELEASE'
xmlUnitVersion = '1.6'
xstreamVersion = '1.4.7'

157
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/DefaultLockRepository.java Normal file
View File

@@ -0,0 +1,157 @@
/*
* Copyright 2016 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
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.jdbc.lock;
import java.util.Date;
import java.util.UUID;
import javax.sql.DataSource;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Repository;
import org.springframework.transaction.annotation.Isolation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.Assert;
/**
* The default implementation of the {@link LockRepository} based on the
* table from the script presented in the {@code org/springframework/integration/jdbc/schema-*.sql}.
* <p>
* This repository can't be shared between different {@link JdbcLockRegistry} instances.
* Otherwise it opens a possibility to break {@link java.util.concurrent.locks.Lock} contract,
* where {@link JdbcLockRegistry} uses non-shared {@link java.util.concurrent.locks.ReentrantLock}s
* for local synchronizations.
*
* @author Dave Syer
* @author Artem Bilan
* @since 4.3
*/
@Repository
@Transactional
public class DefaultLockRepository implements LockRepository, InitializingBean {
/**
* Default value for the table prefix property.
*/
public static final String DEFAULT_TABLE_PREFIX = "INT_";
private final String id = UUID.randomUUID().toString();
private final JdbcTemplate template;
private int ttl = 10000;
private String prefix = DEFAULT_TABLE_PREFIX;
private String region = "DEFAULT";
private String deleteQuery = "DELETE FROM %SLOCK WHERE REGION=? AND LOCK_KEY=? AND CLIENT_ID=?";
private String deleteExpiredQuery = "DELETE FROM %SLOCK WHERE REGION=? AND LOCK_KEY=? AND CREATED_DATE<?";
private String deleteAllQuery = "DELETE FROM %SLOCK WHERE REGION=? AND CLIENT_ID=?";
private String updateQuery = "UPDATE %SLOCK SET CREATED_DATE=? WHERE REGION=? AND LOCK_KEY=? AND CLIENT_ID=?";
private String insertQuery = "INSERT INTO %SLOCK (REGION, LOCK_KEY, CLIENT_ID, CREATED_DATE) VALUES (?, ?, ?, ?)";
private String countQuery = "SELECT COUNT(REGION) FROM %SLOCK WHERE REGION=? AND LOCK_KEY=? AND CLIENT_ID=? AND CREATED_DATE>=?";
@Autowired
DefaultLockRepository(DataSource dataSource) {
this.template = new JdbcTemplate(dataSource);
}
/**
* A unique grouping identifier for all locks persisted with this store. Using
* multiple regions allows the store to be partitioned (if necessary) for different
* purposes. Defaults to <code>DEFAULT</code>.
* @param region the region name to set
*/
public void setRegion(String region) {
Assert.hasText(region, "Region must not be null or empty.");
this.region = region;
}
/**
* Specify the prefix for target data base table used from queries.
* @param prefix the prefix to set (default INT_).
*/
public void setPrefix(String prefix) {
this.prefix = prefix;
}
/**
* Specify the time (in milliseconds) to expire dead locks.
* @param timeToLive the time to expire dead locks.
*/
public void setTimeToLive(int timeToLive) {
this.ttl = timeToLive;
}
@Override
public void afterPropertiesSet() throws Exception {
this.deleteQuery = String.format(this.deleteQuery, this.prefix);
this.deleteExpiredQuery = String.format(this.deleteExpiredQuery, this.prefix);
this.deleteAllQuery = String.format(this.deleteAllQuery, this.prefix);
this.updateQuery = String.format(this.updateQuery, this.prefix);
this.insertQuery = String.format(this.insertQuery, this.prefix);
this.countQuery = String.format(this.countQuery, this.prefix);
}
@Override
public void close() {
this.template.update(this.deleteAllQuery, this.region, this.id);
}
@Override
public void delete(String lock) {
this.template.update(this.deleteQuery, this.region, lock, this.id);
}
@Transactional(isolation = Isolation.SERIALIZABLE, timeout = 1)
@Override
public boolean acquire(String lock) {
deleteExpired(lock);
if (this.template.update(this.updateQuery, new Date(), this.region, lock, this.id) > 0) {
return true;
}
try {
return this.template.update(this.insertQuery, this.region, lock, this.id, new Date()) > 0;
}
catch (DuplicateKeyException e) {
return false;
}
}
@Override
public boolean isAcquired(String lock) {
deleteExpired(lock);
return this.template.queryForObject(this.countQuery, Integer.class, this.region, lock, this.id,
new Date(System.currentTimeMillis() - this.ttl)) == 1;
}
private int deleteExpired(String lock) {
return this.template.update(this.deleteExpiredQuery, this.region, lock,
new Date(System.currentTimeMillis() - this.ttl));
}
}

249
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java Normal file
View File

@@ -0,0 +1,249 @@
/*
* Copyright 2016 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
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.jdbc.lock;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.springframework.integration.support.locks.DefaultLockRegistry;
import org.springframework.integration.support.locks.ExpirableLockRegistry;
import org.springframework.integration.support.locks.LockRegistry;
import org.springframework.integration.util.UUIDConverter;
import org.springframework.transaction.TransactionTimedOutException;
import org.springframework.util.Assert;
/**
*
* A {@link LockRegistry} using a shared database to co-ordinate the locks. Provides the
* same semantics as the {@link DefaultLockRegistry}, but the locks taken will be global,
* as long as the underlying database supports the "serializable" isolation level in its
* transactions.
*
* @author Dave Syer
* @author Artem Bilan
* @since 4.3
*/
public class JdbcLockRegistry implements ExpirableLockRegistry {
private final Map<String, JdbcLock> locks = new HashMap<String, JdbcLock>();
private LockRepository client;
public JdbcLockRegistry(LockRepository client) {
this.client = client;
}
@Override
public Lock obtain(Object lockKey) {
Assert.isInstanceOf(String.class, lockKey);
String path = pathFor((String) lockKey);
JdbcLock lock = this.locks.get(path);
if (lock == null) {
synchronized (this.locks) {
lock = this.locks.get(path);
if (lock == null) {
lock = new JdbcLock(this.client, path);
this.locks.put(path, lock);
}
}
}
return lock;
}
private String pathFor(String input) {
return input == null ? null : UUIDConverter.getUUID(input).toString();
}
@Override
public void expireUnusedOlderThan(long age) {
synchronized (this.locks) {
Iterator<Entry<String, JdbcLock>> iterator = this.locks.entrySet().iterator();
long now = System.currentTimeMillis();
while (iterator.hasNext()) {
Entry<String, JdbcLock> entry = iterator.next();
JdbcLock lock = entry.getValue();
if (now - lock.getLastUsed() > age && !lock.isAcquiredInThisProcess()) {
iterator.remove();
}
}
}
}
private static final class JdbcLock implements Lock {
private final LockRepository mutex;
private final String path;
private volatile long lastUsed = System.currentTimeMillis();
private ReentrantLock delegate = new ReentrantLock();
private JdbcLock(LockRepository client, String path) {
this.mutex = client;
this.path = path;
}
public long getLastUsed() {
return this.lastUsed;
}
@Override
public void lock() {
this.delegate.lock();
try {
while (true) {
try {
while (!doLock()) {
Thread.sleep(100);
}
break;
}
catch (TransactionTimedOutException e) {
// try again
}
catch (InterruptedException e) {
/*
* This method must be uninterruptible so catch and ignore
* interrupts and only break out of the while loop when
* we get the lock.
*/
}
}
}
catch (Exception e) {
this.delegate.unlock();
throw new RuntimeException("Failed to lock mutex at " + this.path, e);
}
}
@Override
public void lockInterruptibly() throws InterruptedException {
this.delegate.lockInterruptibly();
try {
while (true) {
try {
while (!this.doLock()) {
Thread.sleep(100);
if (Thread.currentThread().isInterrupted()) {
throw new InterruptedException();
}
}
break;
}
catch (TransactionTimedOutException e) {
// try again
}
}
}
catch (InterruptedException ie) {
this.delegate.unlock();
throw ie;
}
catch (Exception e) {
this.delegate.unlock();
throw new RuntimeException("Failed to lock mutex at " + this.path, e);
}
}
@Override
public boolean tryLock() {
try {
return tryLock(0, TimeUnit.MICROSECONDS);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
return false;
}
}
@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
long now = System.currentTimeMillis();
if (!this.delegate.tryLock(time, unit)) {
return false;
}
try {
long expire = now + TimeUnit.MILLISECONDS.convert(time, unit);
boolean acquired;
while (true) {
try {
while (!(acquired = doLock()) && System.currentTimeMillis() < expire) {
Thread.sleep(100);
}
if (!acquired) {
this.delegate.unlock();
}
return acquired;
}
catch (TransactionTimedOutException e) {
// try again
}
}
}
catch (Exception e) {
this.delegate.unlock();
throw new RuntimeException("Failed to lock mutex at " + this.path, e);
}
}
private boolean doLock() {
boolean acquired = this.mutex.acquire(this.path);
if (acquired) {
this.lastUsed = System.currentTimeMillis();
}
return acquired;
}
@Override
public void unlock() {
if (!this.delegate.isHeldByCurrentThread()) {
throw new IllegalMonitorStateException("You do not own mutex at " + this.path);
}
if (this.delegate.getHoldCount() > 1) {
this.delegate.unlock();
return;
}
try {
this.mutex.delete(this.path);
}
catch (Exception e) {
throw new RuntimeException("Failed to release mutex at " + this.path, e);
}
finally {
this.delegate.unlock();
}
}
@Override
public Condition newCondition() {
throw new UnsupportedOperationException("Conditions are not supported");
}
public boolean isAcquiredInThisProcess() {
return this.mutex.isAcquired(this.path);
}
}
}

40
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/LockRepository.java Normal file
View File

@@ -0,0 +1,40 @@
/*
* Copyright 2016 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
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.jdbc.lock;
import java.io.Closeable;
/**
* Encapsulation of the SQL shunting that is needed for locks. A {@link JdbcLockRegistry}
* needs a reference to a spring-managed (transactional) client service, so this component
* has to be declared as a bean.
*
* @author Dave Syer
* @since 4.3
*/
public interface LockRepository extends Closeable {
boolean isAcquired(String lock);
void delete(String lock);
boolean acquire(String lock);
@Override
void close();
}

8
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql
View File

@@ -27,3 +27,11 @@ CREATE TABLE INT_MESSAGE_GROUP (
UPDATED_DATE TIMESTAMP DEFAULT NULL,
constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION)
);
CREATE TABLE INT_LOCK (
LOCK_KEY CHAR(36),
REGION VARCHAR(100),
CLIENT_ID CHAR(36),
CREATED_DATE TIMESTAMP NOT NULL,
constraint LOCK_PK primary key (LOCK_KEY, REGION)
);

8
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql
View File

@@ -27,3 +27,11 @@ CREATE TABLE INT_MESSAGE_GROUP (
UPDATED_DATE TIMESTAMP DEFAULT NULL,
constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION)
);
CREATE TABLE INT_LOCK (
LOCK_KEY CHAR(36),
REGION VARCHAR(100),
CLIENT_ID CHAR(36),
CREATED_DATE TIMESTAMP NOT NULL,
constraint LOCK_PK primary key (LOCK_KEY, REGION)
);

1
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-db2.sql
View File

@@ -3,4 +3,5 @@
DROP TABLE INT_MESSAGE ;
DROP TABLE INT_MESSAGE_GROUP ;
DROP TABLE INT_GROUP_TO_MESSAGE ;
DROP TABLE INT_LOCK ;
DROP INDEX INT_MESSAGE_IX1 ;

1
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-derby.sql
View File

@@ -3,4 +3,5 @@
DROP TABLE INT_MESSAGE ;
DROP TABLE INT_MESSAGE_GROUP ;
DROP TABLE INT_GROUP_TO_MESSAGE ;
DROP TABLE INT_LOCK ;
DROP INDEX INT_MESSAGE_IX1 ;

1
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-h2.sql
View File

@@ -3,4 +3,5 @@
DROP TABLE INT_MESSAGE IF EXISTS;
DROP TABLE INT_MESSAGE_GROUP IF EXISTS;
DROP TABLE INT_GROUP_TO_MESSAGE IF EXISTS;
DROP TABLE INT_LOCK IF EXISTS;
DROP INDEX INT_MESSAGE_IX1 IF EXISTS;

1
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-hsqldb.sql
View File

@@ -3,4 +3,5 @@
DROP TABLE INT_MESSAGE IF EXISTS;
DROP TABLE INT_MESSAGE_GROUP IF EXISTS;
DROP TABLE INT_GROUP_TO_MESSAGE IF EXISTS;
DROP TABLE INT_LOCK IF EXISTS;
DROP INDEX INT_MESSAGE_IX1 IF EXISTS;

1
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-mysql-5_6_4.sql
View File

@@ -3,4 +3,5 @@
DROP TABLE IF EXISTS INT_MESSAGE ;
DROP TABLE IF EXISTS INT_MESSAGE_GROUP ;
DROP TABLE IF EXISTS INT_GROUP_TO_MESSAGE ;
DROP TABLE IF EXISTS INT_LOCK ;
DROP INDEX IF EXISTS INT_MESSAGE_IX1 ;

1
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-mysql.sql
View File

@@ -3,4 +3,5 @@
DROP TABLE IF EXISTS INT_MESSAGE ;
DROP TABLE IF EXISTS INT_MESSAGE_GROUP ;
DROP TABLE IF EXISTS INT_GROUP_TO_MESSAGE ;
DROP TABLE IF EXISTS INT_LOCK ;
DROP INDEX IF EXISTS INT_MESSAGE_IX1 ;

1
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-oracle10g.sql
View File

@@ -3,4 +3,5 @@
DROP TABLE INT_MESSAGE ;
DROP TABLE INT_MESSAGE_GROUP ;
DROP TABLE INT_GROUP_TO_MESSAGE ;
DROP TABLE INT_LOCK ;
DROP INDEX INT_MESSAGE_IX1 ;

1
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-postgresql.sql
View File

@@ -3,4 +3,5 @@
DROP TABLE INT_MESSAGE ;
DROP TABLE INT_MESSAGE_GROUP ;
DROP TABLE INT_GROUP_TO_MESSAGE ;
DROP TABLE INT_LOCK ;
DROP INDEX INT_MESSAGE_IX1 ;

1
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sqlserver.sql
View File

@@ -3,4 +3,5 @@
DROP TABLE INT_MESSAGE ;
DROP TABLE INT_MESSAGE_GROUP ;
DROP TABLE INT_GROUP_TO_MESSAGE ;
DROP TABLE INT_LOCK ;
DROP INDEX INT_MESSAGE_IX1 ;

1
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sybase.sql
View File

@@ -3,4 +3,5 @@
DROP TABLE INT_MESSAGE ;
DROP TABLE INT_MESSAGE_GROUP ;
DROP TABLE INT_GROUP_TO_MESSAGE ;
DROP TABLE INT_LOCK ;
DROP INDEX INT_MESSAGE_IX1 ;

8
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql
View File

@@ -27,3 +27,11 @@ CREATE TABLE INT_MESSAGE_GROUP (
UPDATED_DATE TIMESTAMP DEFAULT NULL,
constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION)
);
CREATE TABLE INT_LOCK (
LOCK_KEY CHAR(36),
REGION VARCHAR(100),
CLIENT_ID CHAR(36),
CREATED_DATE TIMESTAMP NOT NULL,
constraint LOCK_PK primary key (LOCK_KEY, REGION)
);

8
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql
View File

@@ -27,3 +27,11 @@ CREATE TABLE INT_MESSAGE_GROUP (
UPDATED_DATE TIMESTAMP DEFAULT NULL,
constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION)
);
CREATE TABLE INT_LOCK (
LOCK_KEY CHAR(36),
REGION VARCHAR(100),
CLIENT_ID CHAR(36),
CREATED_DATE TIMESTAMP NOT NULL,
constraint LOCK_PK primary key (LOCK_KEY, REGION)
);

8
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql-5_6_4.sql
View File

@@ -27,3 +27,11 @@ CREATE TABLE INT_MESSAGE_GROUP (
UPDATED_DATE DATETIME(6) DEFAULT NULL,
constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION)
) ENGINE=InnoDB;
CREATE TABLE INT_LOCK (
LOCK_KEY CHAR(36),
REGION VARCHAR(100),
CLIENT_ID CHAR(36),
CREATED_DATE DATETIME(6) NOT NULL,
constraint LOCK_PK primary key (LOCK_KEY, REGION)
) ENGINE=InnoDB;

8
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql
View File

@@ -27,3 +27,11 @@ CREATE TABLE INT_MESSAGE_GROUP (
UPDATED_DATE DATETIME DEFAULT NULL,
constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION)
) ENGINE=InnoDB;
CREATE TABLE INT_LOCK (
LOCK_KEY CHAR(36),
REGION VARCHAR(100),
CLIENT_ID CHAR(36),
CREATED_DATE DATETIME NOT NULL,
constraint LOCK_PK primary key (LOCK_KEY, REGION)
) ENGINE=InnoDB;

8
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle10g.sql
View File

@@ -27,3 +27,11 @@ CREATE TABLE INT_MESSAGE_GROUP (
UPDATED_DATE TIMESTAMP DEFAULT NULL,
constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION)
);
CREATE TABLE INT_LOCK (
LOCK_KEY CHAR(36),
REGION VARCHAR2(100),
CLIENT_ID CHAR(36),
CREATED_DATE TIMESTAMP NOT NULL,
constraint LOCK_PK primary key (LOCK_KEY, REGION)
);

8
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql
View File

@@ -27,3 +27,11 @@ CREATE TABLE INT_MESSAGE_GROUP (
UPDATED_DATE TIMESTAMP DEFAULT NULL,
constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION)
);
CREATE TABLE INT_LOCK (
LOCK_KEY CHAR(36),
REGION VARCHAR(100),
CLIENT_ID CHAR(36),
CREATED_DATE TIMESTAMP NOT NULL,
constraint LOCK_PK primary key (LOCK_KEY, REGION)
);

8
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql
View File

@@ -27,3 +27,11 @@ CREATE TABLE INT_MESSAGE_GROUP (
UPDATED_DATE DATETIME DEFAULT NULL,
constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION)
);
CREATE TABLE INT_LOCK (
LOCK_KEY CHAR(36),
REGION VARCHAR(100),
CLIENT_ID CHAR(36),
CREATED_DATE DATETIME NOT NULL,
constraint LOCK_PK primary key (LOCK_KEY, REGION)
);

8
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql
View File

@@ -27,3 +27,11 @@ CREATE TABLE INT_MESSAGE_GROUP (
UPDATED_DATE DATETIME DEFAULT NULL,
constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION)
) LOCK DATAROWS;
CREATE TABLE INT_LOCK (
LOCK_KEY CHAR(36),
REGION VARCHAR(100),
CLIENT_ID CHAR(36),
CREATED_DATE DATETIME NOT NULL,
constraint LOCK_PK primary key (LOCK_KEY, REGION)
) LOCK DATAROWS;

1
spring-integration-jdbc/src/main/sql/destroy.sql.vpp
View File

@@ -1,4 +1,5 @@
DROP TABLE $!{IFEXISTSBEFORE} INT_MESSAGE $!{IFEXISTS};
DROP TABLE $!{IFEXISTSBEFORE} INT_MESSAGE_GROUP $!{IFEXISTS};
DROP TABLE $!{IFEXISTSBEFORE} INT_GROUP_TO_MESSAGE $!{IFEXISTS};
DROP TABLE $!{IFEXISTSBEFORE} INT_LOCK $!{IFEXISTS};
DROP INDEX $!{IFEXISTSBEFORE} INT_MESSAGE_IX1 $!{IFEXISTS};

8
spring-integration-jdbc/src/main/sql/schema.sql.vpp
View File

@@ -27,3 +27,11 @@ CREATE TABLE INT_MESSAGE_GROUP (
UPDATED_DATE ${TIMESTAMP} DEFAULT NULL,
constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION)
)#if(${VOODOO}) ${VOODOO}#end;
CREATE TABLE INT_LOCK (
LOCK_KEY CHAR(36),
REGION ${VARCHAR}(100),
CLIENT_ID CHAR(36),
CREATED_DATE ${TIMESTAMP} NOT NULL,
constraint LOCK_PK primary key (LOCK_KEY, REGION)
)#if(${VOODOO}) ${VOODOO}#end;

308
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryDifferentClientTests.java Normal file
View File

@@ -0,0 +1,308 @@
/*
* Copyright 2016 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
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.jdbc.lock;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import javax.sql.DataSource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.StopWatch;
/**
* @author Dave Syer
* @author Artem Bilan
* @since 4.3
*/
@ContextConfiguration("JdbcLockRegistryTests-context.xml")
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext // close at the end after class
public class JdbcLockRegistryDifferentClientTests {
private static Log logger = LogFactory.getLog(JdbcLockRegistryDifferentClientTests.class);
@Autowired
private JdbcLockRegistry registry;
@Autowired
private LockRepository client;
@Autowired
private ConfigurableApplicationContext context;
private AnnotationConfigApplicationContext child;
@Autowired
private DataSource dataSource;
@Before
public void clear() {
this.registry.expireUnusedOlderThan(0);
this.client.close();
this.child = new AnnotationConfigApplicationContext();
this.child.register(DefaultLockRepository.class);
this.child.setParent(this.context);
this.child.refresh();
}
@After
public void close() {
if (this.child != null) {
this.child.close();
}
}
@Test
public void testSecondThreadLoses() throws Exception {
for (int i = 0; i < 100; i++) {
final JdbcLockRegistry registry1 = this.registry;
final JdbcLockRegistry registry2 = this.child.getBean(JdbcLockRegistry.class);
final Lock lock1 = registry1.obtain("foo");
final AtomicBoolean locked = new AtomicBoolean();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
final CountDownLatch latch3 = new CountDownLatch(1);
lock1.lockInterruptibly();
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
Lock lock2 = registry2.obtain("foo");
try {
latch1.countDown();
lock2.lockInterruptibly();
latch2.await(10, TimeUnit.SECONDS);
locked.set(true);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
finally {
lock2.unlock();
latch3.countDown();
}
}
});
assertTrue(latch1.await(10, TimeUnit.SECONDS));
assertFalse(locked.get());
lock1.unlock();
latch2.countDown();
assertTrue(latch3.await(10, TimeUnit.SECONDS));
assertTrue(locked.get());
}
}
@Test
public void testBothLock() throws Exception {
for (int i = 0; i < 100; i++) {
final JdbcLockRegistry registry1 = this.registry;
final JdbcLockRegistry registry2 = this.child.getBean(JdbcLockRegistry.class);
final List<String> locked = new ArrayList<String>();
final CountDownLatch latch = new CountDownLatch(2);
ExecutorService pool = Executors.newFixedThreadPool(2);
pool.execute(new Runnable() {
@Override
public void run() {
Lock lock = registry1.obtain("foo");
try {
lock.lockInterruptibly();
locked.add("1");
latch.countDown();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
finally {
try {
lock.unlock();
}
catch (Exception e) {
// ignore
}
}
}
});
pool.execute(new Runnable() {
@Override
public void run() {
Lock lock = registry2.obtain("foo");
try {
lock.lockInterruptibly();
locked.add("2");
latch.countDown();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
finally {
try {
lock.unlock();
}
catch (Exception e) {
// ignore
}
}
}
});
assertTrue(latch.await(10, TimeUnit.SECONDS));
// eventually they both get the lock and release it
assertTrue(locked.contains("1"));
assertTrue(locked.contains("2"));
}
}
@Test
public void testOnlyOneLock() throws Exception {
for (int i = 0; i < 100; i++) {
final List<String> locked = new ArrayList<String>();
final CountDownLatch latch = new CountDownLatch(20);
ExecutorService pool = Executors.newFixedThreadPool(6);
ArrayList<Callable<Boolean>> tasks = new ArrayList<Callable<Boolean>>();
for (int j = 0; j < 20; j++) {
final DefaultLockRepository client = new DefaultLockRepository(this.dataSource);
client.afterPropertiesSet();
this.context.getAutowireCapableBeanFactory().autowireBean(client);
Callable<Boolean> task = new Callable<Boolean>() {
@Override
public Boolean call() {
Lock lock = new JdbcLockRegistry(client).obtain("foo");
try {
if (locked.isEmpty() && lock.tryLock()) {
if (locked.isEmpty()) {
locked.add("done");
return true;
}
}
}
finally {
try {
lock.unlock();
}
catch (Exception e) {
// ignore
}
latch.countDown();
}
return false;
}
};
tasks.add(task);
}
logger.info("Starting: " + i);
pool.invokeAll(tasks);
assertTrue(latch.await(10, TimeUnit.SECONDS));
// eventually they both get the lock and release it
assertEquals(1, locked.size());
assertTrue(locked.contains("done"));
}
}
@Test
public void testExclusiveAccess() throws Exception {
DefaultLockRepository client1 = new DefaultLockRepository(dataSource);
client1.afterPropertiesSet();
final DefaultLockRepository client2 = new DefaultLockRepository(dataSource);
client2.afterPropertiesSet();
Lock lock1 = new JdbcLockRegistry(client1).obtain("foo");
final BlockingQueue<Integer> data = new LinkedBlockingQueue<Integer>();
final CountDownLatch latch1 = new CountDownLatch(1);
lock1.lockInterruptibly();
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
Lock lock2 = new JdbcLockRegistry(client2).obtain("foo");
try {
latch1.countDown();
StopWatch stopWatch = new StopWatch();
stopWatch.start();
lock2.lockInterruptibly();
stopWatch.stop();
data.add(4);
Thread.sleep(10);
data.add(5);
Thread.sleep(10);
data.add(6);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
finally {
lock2.unlock();
}
}
});
assertTrue(latch1.await(10, TimeUnit.SECONDS));
data.add(1);
Thread.sleep(1000);
data.add(2);
Thread.sleep(1000);
data.add(3);
lock1.unlock();
for (int i = 0; i < 6; i++) {
Integer integer = data.poll(10, TimeUnit.SECONDS);
assertNotNull(integer);
assertEquals(i + 1, integer.intValue());
}
}
}

38
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryTests-context.xml Normal file
View File

@@ -0,0 +1,38 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:jdbc="http://www.springframework.org/schema/jdbc"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:tx="http://www.springframework.org/schema/tx"
xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd">
<jdbc:embedded-database id="dataSource" type="DERBY"/>
<jdbc:initialize-database data-source="dataSource" ignore-failures="ALL">
<jdbc:script location="${int.drop.script}"/>
<jdbc:script location="${int.schema.script}"/>
</jdbc:initialize-database>
<context:property-placeholder location="int-${ENVIRONMENT:derby}.properties"
system-properties-mode="OVERRIDE"
ignore-unresolvable="true"
order="1"/>
<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
<property name="dataSource" ref="dataSource"/>
</bean>
<bean id="lockRegistry" class="org.springframework.integration.jdbc.lock.JdbcLockRegistry">
<constructor-arg name="client" ref="lockClient"/>
</bean>
<bean id="lockClient" class="org.springframework.integration.jdbc.lock.DefaultLockRepository">
<constructor-arg name="dataSource" ref="dataSource"/>
</bean>
<tx:annotation-driven/>
</beans>

287
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryTests.java Normal file
View File

@@ -0,0 +1,287 @@
/*
* Copyright 2016 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
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.jdbc.lock;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Dave Syer
* @since 4.3
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext // close at the end after class
public class JdbcLockRegistryTests {
@Autowired
private JdbcLockRegistry registry;
@Autowired
private LockRepository client;
@Before
public void clear() {
this.registry.expireUnusedOlderThan(0);
this.client.close();
}
@Test
public void testLock() throws Exception {
for (int i = 0; i < 10; i++) {
Lock lock = this.registry.obtain("foo");
lock.lock();
try {
assertEquals(1, TestUtils.getPropertyValue(this.registry, "locks", Map.class).size());
}
finally {
lock.unlock();
}
}
Thread.sleep(10);
this.registry.expireUnusedOlderThan(0);
assertEquals(0, TestUtils.getPropertyValue(this.registry, "locks", Map.class).size());
}
@Test
public void testLockInterruptibly() throws Exception {
for (int i = 0; i < 10; i++) {
Lock lock = this.registry.obtain("foo");
lock.lockInterruptibly();
try {
assertEquals(1, TestUtils.getPropertyValue(this.registry, "locks", Map.class).size());
}
finally {
lock.unlock();
}
}
}
@Test
public void testReentrantLock() throws Exception {
for (int i = 0; i < 10; i++) {
Lock lock1 = this.registry.obtain("foo");
lock1.lock();
try {
Lock lock2 = this.registry.obtain("foo");
assertSame(lock1, lock2);
lock2.lock();
lock2.unlock();
}
finally {
lock1.unlock();
}
}
}
@Test
public void testReentrantLockInterruptibly() throws Exception {
for (int i = 0; i < 10; i++) {
Lock lock1 = this.registry.obtain("foo");
lock1.lockInterruptibly();
try {
Lock lock2 = this.registry.obtain("foo");
assertSame(lock1, lock2);
lock2.lockInterruptibly();
lock2.unlock();
}
finally {
lock1.unlock();
}
}
}
@Test
public void testTwoLocks() throws Exception {
for (int i = 0; i < 10; i++) {
Lock lock1 = this.registry.obtain("foo");
lock1.lockInterruptibly();
try {
Lock lock2 = this.registry.obtain("bar");
assertNotSame(lock1, lock2);
lock2.lockInterruptibly();
lock2.unlock();
}
finally {
lock1.unlock();
}
}
}
@Test
public void testTwoThreadsSecondFailsToGetLock() throws Exception {
final Lock lock1 = this.registry.obtain("foo");
lock1.lockInterruptibly();
final AtomicBoolean locked = new AtomicBoolean();
final CountDownLatch latch = new CountDownLatch(1);
Future<Object> result = Executors.newSingleThreadExecutor().submit(new Callable<Object>() {
@Override
public Object call() throws Exception {
Lock lock2 = JdbcLockRegistryTests.this.registry.obtain("foo");
locked.set(lock2.tryLock(200, TimeUnit.MILLISECONDS));
latch.countDown();
try {
lock2.unlock();
}
catch (Exception e) {
return e;
}
return null;
}
});
assertTrue(latch.await(10, TimeUnit.SECONDS));
assertFalse(locked.get());
lock1.unlock();
Object ise = result.get(10, TimeUnit.SECONDS);
assertThat(ise, instanceOf(IllegalMonitorStateException.class));
assertThat(((Exception) ise).getMessage(), containsString("You do not own"));
}
@Test
public void testTwoThreads() throws Exception {
final Lock lock1 = this.registry.obtain("foo");
final AtomicBoolean locked = new AtomicBoolean();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
final CountDownLatch latch3 = new CountDownLatch(1);
lock1.lockInterruptibly();
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
Lock lock2 = JdbcLockRegistryTests.this.registry.obtain("foo");
try {
latch1.countDown();
lock2.lockInterruptibly();
latch2.await(10, TimeUnit.SECONDS);
locked.set(true);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
finally {
lock2.unlock();
latch3.countDown();
}
}
});
assertTrue(latch1.await(10, TimeUnit.SECONDS));
assertFalse(locked.get());
lock1.unlock();
latch2.countDown();
assertTrue(latch3.await(10, TimeUnit.SECONDS));
assertTrue(locked.get());
}
@Test
public void testTwoThreadsDifferentRegistries() throws Exception {
for (int i = 0; i < 100; i++) {
final JdbcLockRegistry registry1 = new JdbcLockRegistry(this.client);
final JdbcLockRegistry registry2 = new JdbcLockRegistry(this.client);
final Lock lock1 = registry1.obtain("foo");
final AtomicBoolean locked = new AtomicBoolean();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
final CountDownLatch latch3 = new CountDownLatch(1);
lock1.lockInterruptibly();
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
Lock lock2 = registry2.obtain("foo");
try {
latch1.countDown();
lock2.lockInterruptibly();
latch2.await(10, TimeUnit.SECONDS);
locked.set(true);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
finally {
lock2.unlock();
latch3.countDown();
}
}
});
assertTrue(latch1.await(10, TimeUnit.SECONDS));
assertFalse(locked.get());
lock1.unlock();
latch2.countDown();
assertTrue(latch3.await(10, TimeUnit.SECONDS));
assertTrue(locked.get());
}
}
@Test
public void testTwoThreadsWrongOneUnlocks() throws Exception {
final Lock lock = this.registry.obtain("foo");
lock.lockInterruptibly();
final AtomicBoolean locked = new AtomicBoolean();
final CountDownLatch latch = new CountDownLatch(1);
Future<Object> result = Executors.newSingleThreadExecutor().submit(new Callable<Object>() {
@Override
public Object call() throws Exception {
try {
lock.unlock();
}
catch (Exception e) {
latch.countDown();
return e;
}
return null;
}
});
assertTrue(latch.await(10, TimeUnit.SECONDS));
assertFalse(locked.get());
lock.unlock();
Object imse = result.get(10, TimeUnit.SECONDS);
assertThat(imse, instanceOf(IllegalMonitorStateException.class));
assertThat(((Exception) imse).getMessage(), containsString("You do not own"));
}
}

33
src/reference/asciidoc/jdbc.adoc
View File

@@ -974,3 +974,36 @@ a `stored-proc-outbound-gateway`
<int-jdbc:parameter name="ID" expression="payload" />
</int-jdbc:stored-proc-outbound-gateway>
----
[[jdbc-lock-registry]]
=== JDBC Lock Registry
Starting with _version 4.3_, the `JdbcLockRegistry` is available.
Certain components (for example aggregator and resequencer) use a lock obtained from a `LockRegistry` instance to ensure that only one thread is manipulating a group at a time.
The `DefaultLockRegistry` performs this function within a single component; you can now configure an external lock registry on these components.
When used with a shared `MessageGroupStore`, the `JdbcLockRegistry` can be use to provide this functionality across multiple application instances, such that only one instance can manipulate the group at a time.
When a lock is released by a local thread, another local thread will generally be able to acquire the lock immediately.
If a lock is released by a thread using a different registry instance, it can take up to 100ms to acquire the lock.
The `JdbcLockRegistry` is based on the `LockRepository` abstraction, where a `DefaultLockRepository` implementation is present.
The data base schema scripts are located in the `org.springframework.integration.jdbc` package divided to the particular RDBMS vendors.
For example the H2 DDL for lock table looks like:
[source,sql]
----
CREATE TABLE INT_LOCK (
LOCK_KEY CHAR(36),
REGION VARCHAR(100),
CLIENT_ID CHAR(36),
CREATED_DATE TIMESTAMP NOT NULL,
constraint LOCK_PK primary key (LOCK_KEY, REGION)
);
----
The `INT_` can be changed according to the target data base design requirements.
Therefore `prefix` property must be used on the `DefaultLockRepository` bean definition.
Sometimes it happens that one application has moved to the state when it can't release distributed lock - remove the particular record in the data base.
For this purpose such dead locks can be expired by the other application on the next locking invocation.
The `timeToLive` (TTL) option on the `DefaultLockRepository` is provided for this purpose.

5
src/reference/asciidoc/whats-new.adoc
View File

@@ -43,6 +43,11 @@ See <<stream-transformer>> for more information.
A new `IntegrationGraphServer` together with the `IntegrationGraphController` REST service are provided to expose the runtime model of a Spring Integration application as a graph.
See <<integration-graph>> for more information.
==== JDBC Lock Registry
A new `JdbcLockRegistry` is provided for distributed locks shared through the data base table.
See <<jdbc-lock-registry>> for more information.
[[x4.3-general]]
=== General Changes