+ * Experimental - APIs may change. + * + * @author Gary Russell + * @since 1.4.1 + * + */ +public class BatchingRabbitTemplate extends RabbitTemplate { + + private final BatchingStrategy batchingStrategy; + + private final TaskScheduler scheduler; + + private volatile ScheduledFuture scheduledTask; + + /** + * @param batchingStrategy the batching strategy. + * @param scheduler the scheduler. + */ + public BatchingRabbitTemplate(BatchingStrategy batchingStrategy, TaskScheduler scheduler) { + this.batchingStrategy = batchingStrategy; + this.scheduler = scheduler; + } + + @Override + public synchronized void send(String exchange, String routingKey, Message message) throws AmqpException { + if (this.scheduledTask != null) { + this.scheduledTask.cancel(false); + } + MessageBatch batch = this.batchingStrategy.addToBatch(exchange, routingKey, message); + if (batch != null) { + super.send(batch.getExchange(), batch.getRoutingKey(), batch.getMessage()); + } + Date next = this.batchingStrategy.nextRelease(); + if (next != null) { + this.scheduledTask = this.scheduler.schedule(new Runnable() { + + @Override + public void run() { + releaseBatches(); + }}, next); + } + } + + private synchronized void releaseBatches() { + MessageBatch batch; + while ((batch = this.batchingStrategy.releaseBatch()) != null) { + super.send(batch.getExchange(), batch.getRoutingKey(), batch.getMessage()); + } + } + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/support/BatchingStrategy.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/support/BatchingStrategy.java new file mode 100644 index 00000000..8e998667 --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/support/BatchingStrategy.java @@ -0,0 +1,54 @@ +/* + * Copyright 2014 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.amqp.rabbit.core.support; + +import java.util.Date; + +import org.springframework.amqp.core.Message; + +/** + * Strategy for batching messages. The methods will never be called concurrently. + *
+ * Experimental - APIs may change.
+ *
+ * @author Gary Russell
+ * @since 1.4.1
+ *
+ */
+public interface BatchingStrategy {
+
+ /**
+ * Add a message to the batch and optionally release the batch.
+ * @param exchange The exchange.
+ * @param routingKey The routing key.
+ * @param message The message.
+ * @return The batched message ({@link MessageBatch}), or null if not ready to release.
+ */
+ MessageBatch addToBatch(String exchange, String routingKey, Message message);
+
+ /**
+ * @return the date the next scheduled release should run, or null if no data to release.
+ */
+ Date nextRelease();
+
+ /**
+ * Release a batch, perhaps due to a timeout. May be called repeatedly
+ * until {@code null} is returned.
+ * @return The batched message, or null if no batches are ready.
+ */
+ MessageBatch releaseBatch();
+
+}
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/support/MessageBatch.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/support/MessageBatch.java
new file mode 100644
index 00000000..40d54f44
--- /dev/null
+++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/support/MessageBatch.java
@@ -0,0 +1,63 @@
+/*
+ * Copyright 2014 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.amqp.rabbit.core.support;
+
+import org.springframework.amqp.core.Message;
+
+/**
+ * An object encapsulating a {@link Message} containing the batch of messages,
+ * the exchange, and routing key.
+ *
+ * @author Gary Russell
+ * @since 1.4.1
+ *
+ */
+public class MessageBatch {
+
+ private final String exchange;
+
+ private final String routingKey;
+
+ private final Message message;
+
+ public MessageBatch(String exchange, String routingKey, Message message) {
+ this.exchange = exchange;
+ this.routingKey = routingKey;
+ this.message = message;
+ }
+
+ /**
+ * @return the exchange
+ */
+ public String getExchange() {
+ return exchange;
+ }
+
+ /**
+ * @return the routingKey
+ */
+ public String getRoutingKey() {
+ return routingKey;
+ }
+
+ /**
+ * @return the message
+ */
+ public Message getMessage() {
+ return message;
+ }
+
+}
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/support/SimpleBatchingStrategy.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/support/SimpleBatchingStrategy.java
new file mode 100644
index 00000000..164b017a
--- /dev/null
+++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/support/SimpleBatchingStrategy.java
@@ -0,0 +1,138 @@
+/*
+ * Copyright 2014 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.amqp.rabbit.core.support;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.util.Assert;
+
+/**
+ * A simple batching strategy that supports only one exchange/routingKey; includes a batch
+ * size, a batched message size limit and a timeout. The message properties from the first
+ * message in the batch is used in the batch message. Each message is preceded by a 4 byte
+ * length field.
+ *
+ * @author Gary Russell
+ * @since 1.4.1
+ *
+ */
+public class SimpleBatchingStrategy implements BatchingStrategy {
+
+ private final int batchSize;
+
+ private final int bufferLimit;
+
+ private final long timeout;
+
+ private final List
@@ -597,8 +613,31 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
throw new MessageRejectedWhileStoppingException();
}
try {
- invokeListener(channel, message);
- } catch (Throwable ex) {
+ Object batchFormat = message.getMessageProperties().getHeaders().get(MessageProperties.SPRING_BATCH_FORMAT);
+ if (MessageProperties.BATCH_FORMAT_LENGTH_HEADER4.equals(batchFormat) && this.deBatchingEnabled) {
+ ByteBuffer byteBuffer = ByteBuffer.wrap(message.getBody());
+ MessageProperties messageProperties = message.getMessageProperties();
+ messageProperties.getHeaders().remove(MessageProperties.SPRING_BATCH_FORMAT);
+ while (byteBuffer.hasRemaining()) {
+ int length = byteBuffer.getInt();
+ if (length < 0 || length > byteBuffer.remaining()) {
+ throw new ListenerExecutionFailedException("Bad batched message received",
+ new MessageConversionException("Insufficient batch data at offset " + byteBuffer.position()),
+ message);
+ }
+ byte[] body = new byte[length];
+ byteBuffer.get(body);
+ messageProperties.setContentLength(length);
+ // Caveat - shared MessageProperties.
+ Message fragment = new Message(body, messageProperties);
+ invokeListener(channel, fragment);
+ }
+ }
+ else {
+ invokeListener(channel, message);
+ }
+ }
+ catch (Throwable ex) {
handleListenerException(ex);
throw ex;
}
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ConditionalRejectingErrorHandler.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ConditionalRejectingErrorHandler.java
index f5fa6201..568d4eb9 100644
--- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ConditionalRejectingErrorHandler.java
+++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ConditionalRejectingErrorHandler.java
@@ -42,7 +42,7 @@ import org.springframework.util.ErrorHandler;
*/
public class ConditionalRejectingErrorHandler implements ErrorHandler {
- protected static final Log logger = LogFactory.getLog(ConditionalRejectingErrorHandler.class);
+ protected final Log logger = LogFactory.getLog(this.getClass());
private final FatalExceptionStrategy exceptionStrategy;
diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplateTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplateTests.java
new file mode 100644
index 00000000..d4dc330c
--- /dev/null
+++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplateTests.java
@@ -0,0 +1,347 @@
+/*
+ * Copyright 2014 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.amqp.rabbit.core;
+
+import static org.hamcrest.Matchers.containsString;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.logging.Log;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.internal.stubbing.answers.DoesNothing;
+
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageDeliveryMode;
+import org.springframework.amqp.core.MessageListener;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.core.support.BatchingStrategy;
+import org.springframework.amqp.rabbit.core.support.SimpleBatchingStrategy;
+import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.amqp.rabbit.test.BrokerRunning;
+import org.springframework.amqp.rabbit.test.BrokerTestUtils;
+import org.springframework.amqp.utils.test.TestUtils;
+import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+import org.springframework.util.StopWatch;
+
+/**
+ * @author Gary Russell
+ * @since 1.4.1
+ *
+ */
+public class BatchingRabbitTemplateTests {
+
+ private static final String ROUTE = "test.queue";
+
+ @Rule
+ public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(ROUTE);
+
+ private CachingConnectionFactory connectionFactory;
+
+ private ThreadPoolTaskScheduler scheduler;
+
+ @Before
+ public void setup() {
+ this.connectionFactory = new CachingConnectionFactory();
+ this.connectionFactory.setHost("localhost");
+ this.connectionFactory.setPort(BrokerTestUtils.getPort());
+ scheduler = new ThreadPoolTaskScheduler();
+ scheduler.setPoolSize(1);
+ scheduler.initialize();
+ }
+
+ @Test
+ public void testSimpleBatch() throws Exception {
+ BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(2, Integer.MAX_VALUE, 30000);
+ BatchingRabbitTemplate template = new BatchingRabbitTemplate(batchingStrategy, this.scheduler);
+ template.setConnectionFactory(this.connectionFactory);
+ MessageProperties props = new MessageProperties();
+ Message message = new Message("foo".getBytes(), props);
+ template.send("", ROUTE, message);
+ message = new Message("bar".getBytes(), props);
+ template.send("", ROUTE, message);
+ Thread.sleep(100);
+ message = template.receive(ROUTE);
+ assertNotNull(message);
+ assertEquals("\u0000\u0000\u0000\u0003foo\u0000\u0000\u0000\u0003bar", new String(message.getBody()));
+ }
+
+ @Test
+ public void testSimpleBatchTimeout() throws Exception {
+ BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(2, Integer.MAX_VALUE, 50);
+ BatchingRabbitTemplate template = new BatchingRabbitTemplate(batchingStrategy, this.scheduler);
+ template.setConnectionFactory(this.connectionFactory);
+ MessageProperties props = new MessageProperties();
+ Message message = new Message("foo".getBytes(), props);
+ template.send("", ROUTE, message);
+ Thread.sleep(100);
+ message = template.receive(ROUTE);
+ assertNotNull(message);
+ assertEquals("foo", new String(message.getBody()));
+ }
+
+ @Test
+ public void testSimpleBatchTimeoutMultiple() throws Exception {
+ BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(2, Integer.MAX_VALUE, 50);
+ BatchingRabbitTemplate template = new BatchingRabbitTemplate(batchingStrategy, this.scheduler);
+ template.setConnectionFactory(this.connectionFactory);
+ MessageProperties props = new MessageProperties();
+ Message message = new Message("foo".getBytes(), props);
+ template.send("", ROUTE, message);
+ template.send("", ROUTE, message);
+ Thread.sleep(100);
+ message = template.receive(ROUTE);
+ assertNotNull(message);
+ assertEquals("\u0000\u0000\u0000\u0003foo\u0000\u0000\u0000\u0003foo", new String(message.getBody()));
+ }
+
+ @Test
+ public void testSimpleBatchBufferLimit() throws Exception {
+ BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(2, 8, 50);
+ BatchingRabbitTemplate template = new BatchingRabbitTemplate(batchingStrategy, this.scheduler);
+ template.setConnectionFactory(this.connectionFactory);
+ MessageProperties props = new MessageProperties();
+ Message message = new Message("foo".getBytes(), props);
+ template.send("", ROUTE, message);
+ message = new Message("bar".getBytes(), props);
+ template.send("", ROUTE, message);
+ Thread.sleep(100);
+ message = template.receive(ROUTE);
+ assertNotNull(message);
+ assertEquals("foo", new String(message.getBody()));
+ Thread.sleep(100);
+ message = template.receive(ROUTE);
+ assertNotNull(message);
+ assertEquals("bar", new String(message.getBody()));
+ }
+
+ @Test
+ public void testSimpleBatchBufferLimitMultiple() throws Exception {
+ BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(2, 15, 30000);
+ BatchingRabbitTemplate template = new BatchingRabbitTemplate(batchingStrategy, this.scheduler);
+ template.setConnectionFactory(this.connectionFactory);
+ MessageProperties props = new MessageProperties();
+ Message message = new Message("foo".getBytes(), props);
+ template.send("", ROUTE, message);
+ template.send("", ROUTE, message);
+ message = new Message("bar".getBytes(), props);
+ template.send("", ROUTE, message);
+ template.send("", ROUTE, message);
+ Thread.sleep(100);
+ message = template.receive(ROUTE);
+ assertNotNull(message);
+ assertEquals("\u0000\u0000\u0000\u0003foo\u0000\u0000\u0000\u0003foo", new String(message.getBody()));
+ Thread.sleep(100);
+ message = template.receive(ROUTE);
+ assertNotNull(message);
+ assertEquals("\u0000\u0000\u0000\u0003bar\u0000\u0000\u0000\u0003bar", new String(message.getBody()));
+ }
+
+ @Test
+ public void testSimpleBatchBiggerThanBufferLimit() throws Exception {
+ BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(2, 2, 30000);
+ BatchingRabbitTemplate template = new BatchingRabbitTemplate(batchingStrategy, this.scheduler);
+ template.setConnectionFactory(this.connectionFactory);
+ MessageProperties props = new MessageProperties();
+ Message message = new Message("foo".getBytes(), props);
+ template.send("", ROUTE, message);
+ message = new Message("bar".getBytes(), props);
+ template.send("", ROUTE, message);
+ Thread.sleep(100);
+ message = template.receive(ROUTE);
+ assertNotNull(message);
+ assertEquals("foo", new String(message.getBody()));
+ Thread.sleep(100);
+ message = template.receive(ROUTE);
+ assertNotNull(message);
+ assertEquals("bar", new String(message.getBody()));
+ }
+
+ @Test
+ // existing buffered; new message bigger than bufferLimit; released immediately
+ public void testSimpleBatchBiggerThanBufferLimitMultiple() throws Exception {
+ BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(2, 6, 30000);
+ BatchingRabbitTemplate template = new BatchingRabbitTemplate(batchingStrategy, this.scheduler);
+ template.setConnectionFactory(this.connectionFactory);
+ MessageProperties props = new MessageProperties();
+ Message message = new Message("f".getBytes(), props);
+ template.send("", ROUTE, message);
+ message = new Message("bar".getBytes(), props);
+ template.send("", ROUTE, message);
+ Thread.sleep(100);
+ message = template.receive(ROUTE);
+ assertNotNull(message);
+ assertEquals("f", new String(message.getBody()));
+ Thread.sleep(100);
+ message = template.receive(ROUTE);
+ assertNotNull(message);
+ assertEquals("bar", new String(message.getBody()));
+ }
+
+ @Test
+ public void testSimpleBatchTwoEqualBufferLimit() throws Exception {
+ BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(10, 14, 30000);
+ BatchingRabbitTemplate template = new BatchingRabbitTemplate(batchingStrategy, this.scheduler);
+ template.setConnectionFactory(this.connectionFactory);
+ MessageProperties props = new MessageProperties();
+ Message message = new Message("foo".getBytes(), props);
+ template.send("", ROUTE, message);
+ message = new Message("bar".getBytes(), props);
+ template.send("", ROUTE, message);
+ Thread.sleep(100);
+ message = template.receive(ROUTE);
+ assertNotNull(message);
+ assertEquals("\u0000\u0000\u0000\u0003foo\u0000\u0000\u0000\u0003bar", new String(message.getBody()));
+ }
+
+ @Test
+ public void testDebatchByContainer() throws Exception {
+ final List