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

INT-4553: Store-backed QueueChannel improvements

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4553
Fixes https://github.com/spring-projects/spring-integration/issues/2628
Fixes https://github.com/spring-projects/spring-integration/issues/2629

- Avoid `size()` calls on the MGS, use `poll()` instead.
- Optimize the indexes for the `INT_CHANNEL_MESSAGE` table.

Avoid size call when no timeout too.

Polishing - PR Comments

Missed a doc fix

Another missed %PREFIX%

Fix underscores

Polishing; PR comments; make MGQ extendable.

Fix version in doc.

* Polishing `@since`
* Use diamonds whenever it is possible

**Cherry-pick to 5.0.x**

# Conflicts:
#	src/reference/asciidoc/jdbc.adoc
#	src/reference/asciidoc/whats-new.adoc
This commit is contained in:
Gary Russell
2018-11-16 13:24:36 -05:00
committed by Artem Bilan
parent 7f8a81b22f
commit be6498008f
15 changed files with 304 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2018 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.
@@ -66,7 +66,7 @@ public class QueueChannel extends AbstractPollableChannel implements QueueChanne
public QueueChannel(int capacity) {
Assert.isTrue(capacity > 0, "The capacity must be a positive integer. " +
"For a zero-capacity alternative, consider using a 'RendezvousChannel'.");
this.queue = new LinkedBlockingQueue<Message<?>>(capacity);
this.queue = new LinkedBlockingQueue<>(capacity);
}
/**
@@ -75,7 +75,7 @@ public class QueueChannel extends AbstractPollableChannel implements QueueChanne
* unbounded queue may lead to OutOfMemoryErrors.
*/
public QueueChannel() {
this(new LinkedBlockingQueue<Message<?>>());
this(new LinkedBlockingQueue<>());
}
@Override
@@ -116,13 +116,19 @@ public class QueueChannel extends AbstractPollableChannel implements QueueChanne
return ((BlockingQueue<Message<?>>) this.queue).poll(timeout, TimeUnit.MILLISECONDS);
}
else {
long nanos = TimeUnit.MILLISECONDS.toNanos(timeout);
long deadline = System.nanoTime() + nanos;
while (this.queue.size() == 0 && nanos > 0) {
this.queueSemaphore.tryAcquire(nanos, TimeUnit.NANOSECONDS); // NOSONAR - ok to ignore result
nanos = deadline - System.nanoTime();
Message<?> message = this.queue.poll();
if (message == null) {
long nanos = TimeUnit.MILLISECONDS.toNanos(timeout);
long deadline = System.nanoTime() + nanos;
while (message == null && nanos > 0) {
this.queueSemaphore.tryAcquire(nanos, TimeUnit.NANOSECONDS); // NOSONAR ok to ignore result
message = this.queue.poll();
if (message == null) {
nanos = deadline - System.nanoTime();
}
}
}
return this.queue.poll();
return message;
}
}
if (timeout == 0) {
@@ -133,10 +139,12 @@ public class QueueChannel extends AbstractPollableChannel implements QueueChanne
return ((BlockingQueue<Message<?>>) this.queue).take();
}
else {
while (this.queue.size() == 0) {
Message<?> message = this.queue.poll();
while (message == null) {
this.queueSemaphore.tryAcquire(50, TimeUnit.MILLISECONDS);
message = this.queue.poll();
}
return this.queue.poll();
return message;
}
}
catch (InterruptedException e) {
@@ -147,7 +155,7 @@ public class QueueChannel extends AbstractPollableChannel implements QueueChanne
@Override
public List<Message<?>> clear() {
List<Message<?>> clearedMessages = new ArrayList<Message<?>>();
List<Message<?>> clearedMessages = new ArrayList<>();
if (this.queue instanceof BlockingQueue) {
((BlockingQueue<Message<?>>) this.queue).drainTo(clearedMessages);
}
@@ -165,7 +173,7 @@ public class QueueChannel extends AbstractPollableChannel implements QueueChanne
if (selector == null) {
return this.clear();
}
List<Message<?>> purgedMessages = new ArrayList<Message<?>>();
List<Message<?>> purgedMessages = new ArrayList<>();
Object[] array = this.queue.toArray();
for (Object o : array) {
Message<?> message = (Message<?>) o;

56
spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2018 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.
@@ -121,6 +121,42 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
return getMessages().iterator();
}
/**
* Get the store.
* @return the store.
* @since 5.0.11
*/
protected BasicMessageGroupStore getMessageGroupStore() {
return this.messageGroupStore;
}
/**
* Get the store lock.
* @return the lock.
* @since 5.0.11
*/
protected Lock getStoreLock() {
return this.storeLock;
}
/**
* Get the not full condition.
* @return the condition.
* @since 5.0.11
*/
protected Condition getMessageStoreNotFull() {
return this.messageStoreNotFull;
}
/**
* Get the not empty condition.
* @return the condition.
* @since 5.0.11
*/
protected Condition getMessageStoreNotEmpty() {
return this.messageStoreNotEmpty;
}
@Override
public int size() {
return this.messageGroupStore.messageGroupSize(this.groupId);
@@ -156,11 +192,11 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
storeLock.lockInterruptibly();
try {
while (this.size() == 0 && timeoutInNanos > 0) {
message = doPoll();
while (message == null && timeoutInNanos > 0) {
timeoutInNanos = this.messageStoreNotEmpty.awaitNanos(timeoutInNanos);
message = doPoll();
}
message = this.doPoll();
}
finally {
storeLock.unlock();
@@ -196,7 +232,7 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
public int drainTo(Collection<? super Message<?>> collection, int maxElements) {
Assert.notNull(collection, "'collection' must not be null");
int originalSize = collection.size();
ArrayList<Message<?>> list = new ArrayList<Message<?>>();
ArrayList<Message<?>> list = new ArrayList<>();
final Lock storeLock = this.storeLock;
try {
storeLock.lockInterruptibly();
@@ -297,7 +333,7 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
while (this.size() == 0) {
this.messageStoreNotEmpty.await();
}
message = this.doPoll();
message = doPoll();
}
finally {
@@ -306,7 +342,7 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
return message;
}
private Collection<Message<?>> getMessages() {
protected Collection<Message<?>> getMessages() {
return this.messageGroupStore.getMessageGroup(this.groupId).getMessages();
}
@@ -314,7 +350,7 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
* It is assumed that the 'storeLock' is being held by the caller, otherwise
* IllegalMonitorStateException may be thrown
*/
private Message<?> doPoll() {
protected Message<?> doPoll() {
Message<?> message = this.messageGroupStore.pollMessageFromGroup(this.groupId);
this.messageStoreNotFull.signal();
return message;
@@ -323,8 +359,9 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
/**
* It is assumed that the 'storeLock' is being held by the caller, otherwise
* IllegalMonitorStateException may be thrown
* @param message the message to offer.
*/
private boolean doOffer(Message<?> message) {
protected boolean doOffer(Message<?> message) {
boolean offered = false;
if (this.capacity == Integer.MAX_VALUE || this.size() < this.capacity) {
this.messageGroupStore.addMessageToGroup(this.groupId, message);
@@ -333,4 +370,5 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
}
return offered;
}
}

198
spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java
View File

@@ -20,8 +20,12 @@ 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 static org.mockito.BDDMockito.willAnswer;
import static org.mockito.Mockito.spy;
import java.util.ArrayDeque;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -46,147 +50,208 @@ public class QueueChannelTests {
@Test
public void testSimpleSendAndReceive() throws Exception {
final AtomicBoolean messageReceived = new AtomicBoolean(false);
final CountDownLatch latch = new CountDownLatch(1);
final QueueChannel channel = new QueueChannel();
new Thread(() -> {
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(() -> {
Message<?> message = channel.receive();
if (message != null) {
messageReceived.set(true);
latch.countDown();
}
}).start();
assertFalse(messageReceived.get());
channel.send(new GenericMessage<String>("testing"));
latch.await(10000, TimeUnit.MILLISECONDS);
assertTrue(messageReceived.get());
});
channel.send(new GenericMessage<>("testing"));
assertTrue(latch.await(10000, TimeUnit.MILLISECONDS));
exec.shutdownNow();
}
@Test
public void testSimpleSendAndReceiveNonBlockingQueue() throws Exception {
final CountDownLatch latch = new CountDownLatch(1);
final QueueChannel channel = new QueueChannel(new ArrayDeque<>());
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(() -> {
Message<?> message = channel.receive();
if (message != null) {
latch.countDown();
}
});
channel.send(new GenericMessage<>("testing"));
assertTrue(latch.await(10000, TimeUnit.MILLISECONDS));
exec.shutdownNow();
}
@Test
public void testSimpleSendAndReceiveNonBlockingQueueWithTimeout() throws Exception {
final CountDownLatch latch = new CountDownLatch(1);
final QueueChannel channel = new QueueChannel(new ArrayDeque<>());
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(() -> {
Message<?> message = channel.receive(1);
if (message != null) {
latch.countDown();
}
});
channel.send(new GenericMessage<>("testing"));
assertTrue(latch.await(10000, TimeUnit.MILLISECONDS));
exec.shutdownNow();
}
@Test
public void testImmediateReceive() throws Exception {
final AtomicBoolean messageReceived = new AtomicBoolean(false);
final AtomicBoolean messageNull = new AtomicBoolean(false);
final QueueChannel channel = new QueueChannel();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
Runnable receiveTask1 = () -> {
Message<?> message = channel.receive(0);
if (message != null) {
messageReceived.set(true);
}
messageNull.set(message == null);
latch1.countDown();
};
Runnable sendTask = () -> channel.send(new GenericMessage<String>("testing"));
Runnable sendTask = () -> channel.send(new GenericMessage<>("testing"));
singleThreadExecutor.execute(receiveTask1);
latch1.await();
assertTrue(latch1.await(10, TimeUnit.SECONDS));
singleThreadExecutor.execute(sendTask);
assertFalse(messageReceived.get());
Runnable receiveTask2 = () -> {
Message<?> message = channel.receive(0);
if (message != null) {
messageReceived.set(true);
latch2.countDown();
}
latch2.countDown();
};
singleThreadExecutor.execute(receiveTask2);
latch2.await();
assertTrue(messageReceived.get());
assertTrue(latch2.await(10, TimeUnit.SECONDS));
singleThreadExecutor.shutdownNow();
}
@Test
public void testBlockingReceiveWithNoTimeout() throws Exception {
final QueueChannel channel = new QueueChannel();
final AtomicBoolean receiveInterrupted = new AtomicBoolean(false);
final AtomicBoolean messageNull = new AtomicBoolean(false);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(() -> {
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(() -> {
Message<?> message = channel.receive();
receiveInterrupted.set(true);
assertTrue(message == null);
messageNull.set(message == null);
latch.countDown();
});
t.start();
assertFalse(receiveInterrupted.get());
t.interrupt();
latch.await();
assertTrue(receiveInterrupted.get());
exec.shutdownNow();
assertTrue(latch.await(10, TimeUnit.SECONDS));
assertTrue(messageNull.get());
}
@Test
public void testBlockingReceiveWithTimeout() throws Exception {
final QueueChannel channel = new QueueChannel();
final AtomicBoolean receiveInterrupted = new AtomicBoolean(false);
final AtomicBoolean messageNull = new AtomicBoolean(false);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(() -> {
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(() -> {
Message<?> message = channel.receive(10000);
receiveInterrupted.set(true);
assertTrue(message == null);
messageNull.set(message == null);
latch.countDown();
});
t.start();
assertFalse(receiveInterrupted.get());
t.interrupt();
latch.await();
assertTrue(receiveInterrupted.get());
exec.shutdownNow();
assertTrue(latch.await(10, TimeUnit.SECONDS));
assertTrue(messageNull.get());
}
@Test
public void testBlockingReceiveWithTimeoutEmptyThenSend() throws Exception {
Queue<Message<?>> queue = spy(new ArrayDeque<>());
CountDownLatch pollLatch = new CountDownLatch(1);
AtomicBoolean first = new AtomicBoolean(true);
willAnswer(i -> {
pollLatch.countDown();
return first.getAndSet(false) ? null : i.callRealMethod();
}).given(queue).poll();
final QueueChannel channel = new QueueChannel(queue);
final CountDownLatch latch = new CountDownLatch(1);
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(() -> {
Message<?> message = channel.receive(10000);
if (message != null) {
latch.countDown();
}
});
assertTrue(pollLatch.await(10, TimeUnit.SECONDS));
channel.send(new GenericMessage<>("testing"));
assertTrue(latch.await(10, TimeUnit.SECONDS));
exec.shutdownNow();
}
@Test
public void testBlockingReceiveNoTimeoutEmptyThenSend() throws Exception {
Queue<Message<?>> queue = spy(new ArrayDeque<>());
CountDownLatch pollLatch = new CountDownLatch(1);
AtomicBoolean first = new AtomicBoolean(true);
willAnswer(i -> {
pollLatch.countDown();
return first.getAndSet(false) ? null : i.callRealMethod();
}).given(queue).poll();
final QueueChannel channel = new QueueChannel(queue);
final CountDownLatch latch = new CountDownLatch(1);
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(() -> {
Message<?> message = channel.receive();
if (message != null) {
latch.countDown();
}
});
assertTrue(pollLatch.await(10, TimeUnit.SECONDS));
channel.send(new GenericMessage<>("testing"));
assertTrue(latch.await(10, TimeUnit.SECONDS));
exec.shutdownNow();
}
@Test
public void testImmediateSend() {
QueueChannel channel = new QueueChannel(3);
boolean result1 = channel.send(new GenericMessage<String>("test-1"));
boolean result1 = channel.send(new GenericMessage<>("test-1"));
assertTrue(result1);
boolean result2 = channel.send(new GenericMessage<String>("test-2"), 100);
boolean result2 = channel.send(new GenericMessage<>("test-2"), 100);
assertTrue(result2);
boolean result3 = channel.send(new GenericMessage<String>("test-3"), 0);
boolean result3 = channel.send(new GenericMessage<>("test-3"), 0);
assertTrue(result3);
boolean result4 = channel.send(new GenericMessage<String>("test-4"), 0);
boolean result4 = channel.send(new GenericMessage<>("test-4"), 0);
assertFalse(result4);
}
@Test
public void testBlockingSendWithNoTimeout() throws Exception {
final QueueChannel channel = new QueueChannel(1);
boolean result1 = channel.send(new GenericMessage<String>("test-1"));
boolean result1 = channel.send(new GenericMessage<>("test-1"));
assertTrue(result1);
final AtomicBoolean sendInterrupted = new AtomicBoolean(false);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(() -> {
channel.send(new GenericMessage<String>("test-2"));
sendInterrupted.set(true);
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(() -> {
channel.send(new GenericMessage<>("test-2"));
latch.countDown();
});
t.start();
assertFalse(sendInterrupted.get());
t.interrupt();
latch.await();
assertTrue(sendInterrupted.get());
exec.shutdownNow();
assertTrue(latch.await(10, TimeUnit.SECONDS));
}
@Test
public void testBlockingSendWithTimeout() throws Exception {
final QueueChannel channel = new QueueChannel(1);
boolean result1 = channel.send(new GenericMessage<String>("test-1"));
boolean result1 = channel.send(new GenericMessage<>("test-1"));
assertTrue(result1);
final AtomicBoolean sendInterrupted = new AtomicBoolean(false);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(() -> {
channel.send(new GenericMessage<String>("test-2"), 10000);
sendInterrupted.set(true);
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(() -> {
channel.send(new GenericMessage<>("test-2"), 10000);
latch.countDown();
});
t.start();
assertFalse(sendInterrupted.get());
t.interrupt();
latch.await();
assertTrue(sendInterrupted.get());
exec.shutdownNow();
assertTrue(latch.await(10, TimeUnit.SECONDS));
}
@Test
public void testClear() {
QueueChannel channel = new QueueChannel(2);
GenericMessage<String> message1 = new GenericMessage<String>("test1");
GenericMessage<String> message2 = new GenericMessage<String>("test2");
GenericMessage<String> message3 = new GenericMessage<String>("test3");
GenericMessage<String> message1 = new GenericMessage<>("test1");
GenericMessage<String> message2 = new GenericMessage<>("test2");
GenericMessage<String> message3 = new GenericMessage<>("test3");
assertTrue(channel.send(message1));
assertTrue(channel.send(message2));
assertFalse(channel.send(message3, 0));
@@ -217,11 +282,11 @@ public class QueueChannelTests {
.setExpirationDate(future).build();
assertTrue(channel.send(expiredMessage, 0));
assertTrue(channel.send(unexpiredMessage, 0));
assertFalse(channel.send(new GenericMessage<String>("atCapacity"), 0));
assertFalse(channel.send(new GenericMessage<>("atCapacity"), 0));
List<Message<?>> purgedMessages = channel.purge(new UnexpiredMessageSelector());
assertNotNull(purgedMessages);
assertEquals(1, purgedMessages.size());
assertTrue(channel.send(new GenericMessage<String>("roomAvailable"), 0));
assertTrue(channel.send(new GenericMessage<>("roomAvailable"), 0));
}
@Rule
@@ -340,4 +405,5 @@ public class QueueChannelTests {
assertTrue(latch4.await(1000, TimeUnit.MILLISECONDS));
}
*/
}

65
spring-integration-core/src/test/java/org/springframework/integration/store/MessageGroupQueueTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2018 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.
@@ -19,15 +19,20 @@ package org.springframework.integration.store;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.BDDMockito.willAnswer;
import static org.mockito.Mockito.spy;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.CompletionService;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -38,20 +43,56 @@ import org.springframework.messaging.support.GenericMessage;
/**
* @author Dave Syer
* @author Gary Russell
* @since 2.0
*/
public class MessageGroupQueueTests {
static final Log logger = LogFactory.getLog(MessageGroupQueueTests.class);
private static final Log logger = LogFactory.getLog(MessageGroupQueueTests.class);
@Test
public void testPutAndPoll() throws Exception {
MessageGroupQueue queue = new MessageGroupQueue(new SimpleMessageStore(), "FOO");
queue.put(new GenericMessage<String>("foo"));
queue.put(new GenericMessage<>("foo"));
Message<?> result = queue.poll(100, TimeUnit.MILLISECONDS);
assertNotNull(result);
}
@Test
public void testPollTimeout() throws Exception {
MessageGroupQueue queue = new MessageGroupQueue(new SimpleMessageStore(), "FOO");
Message<?> result = queue.poll(1, TimeUnit.MILLISECONDS);
assertNull(result);
}
@Test
public void testPollEmpty() throws Exception {
MessageGroupQueue queue = spy(new MessageGroupQueue(new SimpleMessageStore(), "FOO"));
CountDownLatch latch1 = new CountDownLatch(1);
AtomicBoolean first = new AtomicBoolean(true);
willAnswer(i -> {
latch1.countDown();
return first.getAndSet(false) ? null : i.callRealMethod();
}).given(queue).doPoll();
ExecutorService exec = Executors.newSingleThreadExecutor();
CountDownLatch latch2 = new CountDownLatch(1);
exec.execute(() -> {
try {
Message<?> result = queue.poll(100, TimeUnit.MILLISECONDS);
if (result != null) {
latch2.countDown();
}
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
assertTrue(latch1.await(10, TimeUnit.SECONDS));
queue.put(new GenericMessage<>("foo"));
assertTrue(latch2.await(10, TimeUnit.SECONDS));
exec.shutdownNow();
}
@Test
public void testSize() throws Exception {
MessageGroupQueue queue = new MessageGroupQueue(new SimpleMessageStore(), "FOO");
@@ -65,9 +106,9 @@ public class MessageGroupQueueTests {
public void testCapacityAfterExpiry() throws Exception {
SimpleMessageStore messageGroupStore = new SimpleMessageStore();
MessageGroupQueue queue = new MessageGroupQueue(messageGroupStore, "FOO", 2);
queue.put(new GenericMessage<String>("foo"));
queue.put(new GenericMessage<>("foo"));
assertEquals(1, queue.remainingCapacity());
queue.put(new GenericMessage<String>("bar"));
queue.put(new GenericMessage<>("bar"));
assertEquals(0, queue.remainingCapacity());
Message<?> result = queue.poll(100, TimeUnit.MILLISECONDS);
assertNotNull(result);
@@ -78,21 +119,21 @@ public class MessageGroupQueueTests {
public void testCapacityExceeded() throws Exception {
SimpleMessageStore messageGroupStore = new SimpleMessageStore();
MessageGroupQueue queue = new MessageGroupQueue(messageGroupStore, "FOO", 1);
queue.put(new GenericMessage<String>("foo"));
assertFalse(queue.offer(new GenericMessage<String>("bar"), 100, TimeUnit.MILLISECONDS));
queue.put(new GenericMessage<>("foo"));
assertFalse(queue.offer(new GenericMessage<>("bar"), 100, TimeUnit.MILLISECONDS));
}
@Test
public void testPutAndTake() throws Exception {
MessageGroupQueue queue = new MessageGroupQueue(new SimpleMessageStore(), "FOO");
queue.put(new GenericMessage<String>("foo"));
queue.put(new GenericMessage<>("foo"));
Message<?> result = queue.take();
assertNotNull(result);
}
@Test
public void testConcurrentAccess() throws Exception {
doTestConcurrentAccess(50, 20, new HashSet<String>());
doTestConcurrentAccess(50, 20, new HashSet<>());
}
@Test
@@ -101,11 +142,10 @@ public class MessageGroupQueueTests {
}
private void doTestConcurrentAccess(int concurrency, final int maxPerTask, final Set<String> set) throws Exception {
SimpleMessageStore messageGroupStore = new SimpleMessageStore();
final MessageGroupQueue queue = new MessageGroupQueue(messageGroupStore, "FOO");
ExecutorService executorService = Executors.newCachedThreadPool();
CompletionService<Boolean> completionService = new ExecutorCompletionService<Boolean>(executorService);
CompletionService<Boolean> completionService = new ExecutorCompletionService<>(executorService);
for (int i = 0; i < concurrency; i++) {
@@ -114,7 +154,7 @@ public class MessageGroupQueueTests {
completionService.submit(() -> {
boolean result = true;
for (int j = 0; j < maxPerTask; j++) {
result &= queue.add(new GenericMessage<String>("count=" + big + ":" + j));
result &= queue.add(new GenericMessage<>("count=" + big + ":" + j));
if (!result) {
logger.warn("Failed to add");
}
@@ -158,7 +198,6 @@ public class MessageGroupQueueTests {
assertEquals(Integer.MAX_VALUE, queue.remainingCapacity());
executorService.shutdown();
}
}

10
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/OracleChannelMessageStoreQueryProvider.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2018 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.
@@ -27,6 +27,8 @@ import org.springframework.jdbc.core.JdbcTemplate;
*
* @author Gunnar Hillert
* @author Artem Bilan
* @author Gary Russell
*
* @since 2.2
*/
public class OracleChannelMessageStoreQueryProvider extends AbstractChannelMessageStoreQueryProvider {
@@ -53,7 +55,8 @@ public class OracleChannelMessageStoreQueryProvider extends AbstractChannelMessa
@Override
public String getPriorityPollFromGroupExcludeIdsQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
return "SELECT /*+ INDEX(%PREFIX%CHANNEL_MESSAGE %PREFIX%CHANNEL_MSG_PRIORITY_IDX) */ " +
"%PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) " +
"order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE FOR UPDATE SKIP LOCKED";
@@ -61,7 +64,8 @@ public class OracleChannelMessageStoreQueryProvider extends AbstractChannelMessa
@Override
public String getPriorityPollFromGroupQuery() {
return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
return "SELECT /*+ INDEX(%PREFIX%CHANNEL_MESSAGE %PREFIX%CHANNEL_MSG_PRIORITY_IDX) */ " +
"%PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " +
"where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " +
"order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE FOR UPDATE SKIP LOCKED";
}

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

@@ -46,11 +46,12 @@ CREATE TABLE INT_CHANNEL_MESSAGE (
MESSAGE_SEQUENCE BIGINT NOT NULL ,
MESSAGE_BYTES BLOB,
REGION VARCHAR(100) NOT NULL,
constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
constraint INT_CHANNEL_MESSAGE_PK primary key (REGION, GROUP_KEY, CREATED_DATE, MESSAGE_SEQUENCE)
);
CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE);
CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE);
CREATE INDEX INT_CHANNEL_MSG_DELETE_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_ID);
-- This is only needed if the message group store property 'priorityEnabled' is true
-- CREATE UNIQUE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE);
CREATE TABLE INT_METADATA_STORE (

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

@@ -46,11 +46,12 @@ CREATE TABLE INT_CHANNEL_MESSAGE (
MESSAGE_SEQUENCE BIGINT NOT NULL GENERATED ALWAYS AS IDENTITY (START WITH 1, INCREMENT BY 1),
MESSAGE_BYTES BLOB,
REGION VARCHAR(100) NOT NULL,
constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
constraint INT_CHANNEL_MESSAGE_PK primary key (REGION, GROUP_KEY, CREATED_DATE, MESSAGE_SEQUENCE)
);
CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE);
CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE);
CREATE INDEX INT_CHANNEL_MSG_DELETE_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_ID);
-- This is only needed if the message group store property 'priorityEnabled' is true
-- CREATE UNIQUE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE);
CREATE TABLE INT_METADATA_STORE (

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

@@ -46,11 +46,12 @@ CREATE TABLE INT_CHANNEL_MESSAGE (
MESSAGE_SEQUENCE BIGINT NOT NULL ,
MESSAGE_BYTES LONGVARBINARY,
REGION VARCHAR(100) NOT NULL,
constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
constraint INT_CHANNEL_MESSAGE_PK primary key (REGION, GROUP_KEY, CREATED_DATE, MESSAGE_SEQUENCE)
);
CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE);
CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE);
CREATE INDEX INT_CHANNEL_MSG_DELETE_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_ID);
-- This is only needed if the message group store property 'priorityEnabled' is true
-- CREATE UNIQUE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE);
CREATE TABLE INT_METADATA_STORE (

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

@@ -46,11 +46,12 @@ CREATE TABLE INT_CHANNEL_MESSAGE (
MESSAGE_SEQUENCE BIGINT NOT NULL ,
MESSAGE_BYTES LONGVARBINARY,
REGION VARCHAR(100) NOT NULL,
constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
constraint INT_CHANNEL_MESSAGE_PK primary key (REGION, GROUP_KEY, CREATED_DATE, MESSAGE_SEQUENCE)
);
CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE);
CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE);
CREATE INDEX INT_CHANNEL_MSG_DELETE_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_ID);
-- This is only needed if the message group store property 'priorityEnabled' is true
-- CREATE UNIQUE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE);
CREATE TABLE INT_METADATA_STORE (

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

@@ -46,11 +46,12 @@ CREATE TABLE INT_CHANNEL_MESSAGE (
MESSAGE_SEQUENCE BIGINT NOT NULL AUTO_INCREMENT UNIQUE,
MESSAGE_BYTES BLOB,
REGION VARCHAR(100) NOT NULL,
constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
constraint INT_CHANNEL_MESSAGE_PK primary key (REGION, GROUP_KEY, CREATED_DATE, MESSAGE_SEQUENCE)
) ENGINE=InnoDB;
CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE);
CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE);
CREATE INDEX INT_CHANNEL_MSG_DELETE_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_ID);
-- This is only needed if the message group store property 'priorityEnabled' is true
-- CREATE UNIQUE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE);
CREATE TABLE INT_METADATA_STORE (

7
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle.sql
View File

@@ -46,11 +46,12 @@ CREATE TABLE INT_CHANNEL_MESSAGE (
MESSAGE_SEQUENCE NUMBER(19,0) NOT NULL ,
MESSAGE_BYTES BLOB,
REGION VARCHAR2(100) NOT NULL,
constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
constraint INT_CHANNEL_MESSAGE_PK primary key (REGION, GROUP_KEY, CREATED_DATE, MESSAGE_SEQUENCE)
);
CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE);
CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE);
CREATE INDEX INT_CHANNEL_MSG_DELETE_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_ID);
-- This is only needed if the message group store property 'priorityEnabled' is true
-- CREATE UNIQUE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE);
CREATE TABLE INT_METADATA_STORE (

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

@@ -46,11 +46,12 @@ CREATE TABLE INT_CHANNEL_MESSAGE (
MESSAGE_SEQUENCE BIGINT NOT NULL DEFAULT nextval('INT_MESSAGE_SEQ'),
MESSAGE_BYTES BYTEA,
REGION VARCHAR(100) NOT NULL,
constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
constraint INT_CHANNEL_MESSAGE_PK primary key (REGION, GROUP_KEY, CREATED_DATE, MESSAGE_SEQUENCE)
);
CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE);
CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE);
CREATE INDEX INT_CHANNEL_MSG_DELETE_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_ID);
-- This is only needed if the message group store property 'priorityEnabled' is true
-- CREATE UNIQUE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE);
CREATE TABLE INT_METADATA_STORE (

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

@@ -46,11 +46,12 @@ CREATE TABLE INT_CHANNEL_MESSAGE (
MESSAGE_SEQUENCE BIGINT NOT NULL ,
MESSAGE_BYTES IMAGE,
REGION VARCHAR(100) NOT NULL,
constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
constraint INT_CHANNEL_MESSAGE_PK primary key (REGION, GROUP_KEY, CREATED_DATE, MESSAGE_SEQUENCE)
);
CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE);
CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE);
CREATE INDEX INT_CHANNEL_MSG_DELETE_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_ID);
-- This is only needed if the message group store property 'priorityEnabled' is true
-- CREATE UNIQUE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE);
CREATE TABLE INT_METADATA_STORE (

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

@@ -46,11 +46,12 @@ CREATE TABLE INT_CHANNEL_MESSAGE (
MESSAGE_SEQUENCE BIGINT NOT NULL ,
MESSAGE_BYTES IMAGE,
REGION VARCHAR(100) NOT NULL,
constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
constraint INT_CHANNEL_MESSAGE_PK primary key (REGION, GROUP_KEY, CREATED_DATE, MESSAGE_SEQUENCE)
) LOCK DATAROWS;
CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE);
CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE);
CREATE INDEX INT_CHANNEL_MSG_DELETE_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_ID);
-- This is only needed if the message group store property 'priorityEnabled' is true
-- CREATE UNIQUE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE);
CREATE TABLE INT_METADATA_STORE (

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

@@ -153,8 +153,8 @@ If not explicitly set, the attribute defaults to _0_.
[[jdbc-outbound-channel-adapter]]
=== Outbound Channel Adapter
The outbound Channel Adapter is the inverse of the inbound: its role is to handle a message and use it to execute a SQL query.
The message payload and headers are available by default as input parameters to the query, for instance:
The outbound channel adapter is the inverse of the inbound: its role is to handle a message and use it to execute a SQL query.
By default, the message payload and headers are available as input parameters to the query, as the following example shows:
[source,xml]
----
@@ -269,7 +269,7 @@ For example:
[source,xml]
----
<int-jdbc:outbound-gateway
update="insert into foos (status, name) values (0, :payload[foo])"
update="insert into mythings (status, name) values (0, :payload[thing])"
request-channel="input" reply-channel="output" data-source="dataSource"
keys-generated="true"/>
----
@@ -327,6 +327,12 @@ Spring Integration provides 2 JDBC specific Message Store implementations.
The first one, is the `JdbcMessageStore` which is suitable to be used in conjunction with _Aggregators_ and the _Claim-Check_ pattern.
While it can be used for backing _Message Channels_ as well, you may want to consider using the `JdbcChannelMessageStore` implementation instead, as it provides a more targeted and scalable implementation.
IMPORTANT: Starting with versions 5.0.11, 5.1.2, the indexes for the `JdbcChannelMessageStore` have been optimized.
If you have large message groups in such a store, you may wish to alter the indexes.
Furthermore, the index for `PriorityChannel` is commented out because it is not needed unless you are using such channels backed by JDBC.
NOTE: When using the `OracleChannelMessageStoreQueryProvider`, the priority channel index **must** be added because it is included in a hint in the query.
==== Initializing the Database
Before starting to use JDBC Message Store components, it is important to provision target data base with the appropriate objects.