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

AMQP-422 PublisherCallbackChannel: Postpone Close

Browse Source 
JIRA: https://jira.spring.io/browse/AMQP-422

Add `stopSemaphore` and `closeTimeout` option to the `PublisherCallbackChannelImpl`
to postpone physical `close` until timeout or handlong for all pending confirms

Decrease `@Repeat` for `RabbitTemplatePerformanceIntegrationTests` for better build performance

AMQP-422: Rework `stop` logic to the scheduling

AMQP-422: use `waitForConfirmsOrDie` from CCF

* Revert all changes to the `PublisherCallbackChannelImpl` class (leave just code style polishing)
* Add to the `CachingConnectionFactory.CachedChannelInvocationHandler.physicalClose()` to postpone
`Channel.close()` `if(publisherConfirms)` and use `waitForConfirmsOrDie` in that separate `Thread`
* Add a couple of test to demonstrate new logic
* in case of just `publisherReturns` `Channel` is closed normally as it is now

AMQP-422: Add `active` check to the async close logic

AMQP-422: NPE fix and `catch` logic

AMQP-422: Accept stop schedule for `publisherReturns` too

Change wait timeout to 5 seconds

AMQP-422: Use `Thread.sleep(5000);` for `publisherReturns`

Polishing
This commit is contained in:
Artem Bilan
2014-10-01 14:13:20 +03:00
committed by Gary Russell
parent 5906c1e246
commit aaf9cffbae
6 changed files with 140 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java
View File

@@ -175,6 +175,10 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di
}
}
protected ExecutorService getExecutorService() {
return executorService;
}
/**
* How long to wait (milliseconds) for a response to a connection close
* operation from the broker; default 30000 (30 seconds).
@@ -185,6 +189,10 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di
this.closeTimeout = closeTimeout;
}
public int getCloseTimeout() {
return closeTimeout;
}
protected final Connection createBareConnection() {
try {
if (this.addresses != null) {

49
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java
View File

@@ -25,6 +25,8 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -115,6 +117,10 @@ public class CachingConnectionFactory extends AbstractConnectionFactory implemen
/** Synchronization monitor for the shared Connection */
private final Object connectionMonitor = new Object();
/** Executor used for deferred close if no explicit executor set. */
private final ExecutorService deferredCloseExecutor = Executors.newCachedThreadPool();
/**
* Create a new CachingConnectionFactory initializing the hostname to be the value returned from
* InetAddress.getLocalHost(), or "localhost" if getLocalHost() throws an exception.
@@ -612,7 +618,45 @@ public class CachingConnectionFactory extends AbstractConnectionFactory implemen
return;
}
try {
this.target.close();
if (CachingConnectionFactory.this.active &&
(CachingConnectionFactory.this.publisherConfirms ||
CachingConnectionFactory.this.publisherReturns)) {
ExecutorService executorService = (getExecutorService() != null
? getExecutorService()
: CachingConnectionFactory.this.deferredCloseExecutor);
final Channel channel = CachedChannelInvocationHandler.this.target;
executorService.execute(new Runnable() {
@Override
public void run() {
try {
if (CachingConnectionFactory.this.publisherConfirms) {
channel.waitForConfirmsOrDie(5000);
}
else {
Thread.sleep(5000);
}
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
catch (Exception e) {}
finally {
try {
if (channel.isOpen()) {
channel.close();
}
}
catch (IOException e) {}
catch (AlreadyClosedException e) {}
}
}
});
}
else {
this.target.close();
}
}
catch (AlreadyClosedException e) {
if (logger.isTraceEnabled()) {
@@ -642,8 +686,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory implemen
@Override
public Channel createChannel(boolean transactional) {
Channel channel = getChannel(this, transactional);
return channel;
return getChannel(this, transactional);
}
@Override

20
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java
View File

@@ -73,15 +73,19 @@ import com.rabbitmq.client.ShutdownSignalException;
* @since 1.0.1
*
*/
public class PublisherCallbackChannelImpl implements PublisherCallbackChannel, ConfirmListener, ReturnListener, ShutdownListener {
public class PublisherCallbackChannelImpl
implements PublisherCallbackChannel, ConfirmListener, ReturnListener, ShutdownListener {
private static final String[] METHODS_OF_INTEREST = new String[] {"getFlow", "flow", "flowBlocked", "basicConsume", "basicQos"};
private static final String[] METHODS_OF_INTEREST =
new String[] {"getFlow", "flow", "flowBlocked", "basicConsume", "basicQos"};
private static final MethodFilter METHOD_FILTER = new MethodFilter() {
@Override
public boolean matches(java.lang.reflect.Method method) {
return ObjectUtils.containsElement(METHODS_OF_INTEREST, method.getName());
}
};
private final Log logger = LogFactory.getLog(this.getClass());
@@ -109,7 +113,7 @@ public class PublisherCallbackChannelImpl implements PublisherCallbackChannel, C
delegate.addShutdownListener(this);
this.delegate = delegate;
// The following reflection is required to maintain comatibility with pre 3.3.x clients.
// The following reflection is required to maintain compatibility with pre 3.3.x clients.
final AtomicReference<java.lang.reflect.Method> getFlowMethod = new AtomicReference<java.lang.reflect.Method>();
final AtomicReference<java.lang.reflect.Method> flowMethod = new AtomicReference<java.lang.reflect.Method>();
final AtomicReference<java.lang.reflect.Method> flowBlockedMethod = new AtomicReference<java.lang.reflect.Method>();
@@ -454,7 +458,8 @@ public class PublisherCallbackChannelImpl implements PublisherCallbackChannel, C
return (String) ReflectionUtils.invokeMethod(this.basicConsumeFourArgsMethod, this.delegate, queue,
autoAck, arguments, callback);
}
throw new UnsupportedOperationException("'basicConsume(String, boolean, Map, Consumer)' is not supported by the client library");
throw new UnsupportedOperationException("'basicConsume(String, boolean, Map, Consumer)' " +
"is not supported by the client library");
}
public String basicConsume(String queue, boolean autoAck,
@@ -719,7 +724,7 @@ public class PublisherCallbackChannelImpl implements PublisherCallbackChannel, C
AMQP.BasicProperties properties,
byte[] body) throws IOException
{
Object uuidObject = properties.getHeaders().get(RETURN_CORRELATION).toString();
String uuidObject = properties.getHeaders().get(RETURN_CORRELATION).toString();
Listener listener = this.listeners.get(uuidObject);
if (listener == null || !listener.isReturnListener()) {
if (logger.isWarnEnabled()) {
@@ -748,10 +753,7 @@ public class PublisherCallbackChannelImpl implements PublisherCallbackChannel, C
@Override
public boolean equals(Object obj) {
if (obj == this) {
return true;
}
return this.delegate.equals(obj);
return obj == this || this.delegate.equals(obj);
}
@Override

6
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePerformanceIntegrationTests.java
View File

@@ -88,7 +88,7 @@ public class RabbitTemplatePerformanceIntegrationTests {
}
@Test
@Repeat(2000)
@Repeat(200)
public void testSendAndReceive() throws Exception {
template.convertAndSend(ROUTE, "message");
String result = (String) template.receiveAndConvert(ROUTE);
@@ -104,7 +104,7 @@ public class RabbitTemplatePerformanceIntegrationTests {
}
@Test
@Repeat(2000)
@Repeat(200)
public void testSendAndReceiveTransacted() throws Exception {
template.setChannelTransacted(true);
template.convertAndSend(ROUTE, "message");
@@ -113,7 +113,7 @@ public class RabbitTemplatePerformanceIntegrationTests {
}
@Test
@Repeat(2000)
@Repeat(200)
public void testSendAndReceiveExternalTransacted() throws Exception {
template.setChannelTransacted(true);
new TransactionTemplate(new TestTransactionManager()).execute(new TransactionCallback<Void>() {

67
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java
View File

@@ -74,11 +74,11 @@ import org.springframework.expression.spel.standard.SpelExpressionParser;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
/**
* @author Gary Russell
* @author Gunar Hillert
* @author Artem Bilan
* @since 1.1
*
*/
@@ -460,14 +460,12 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
ccf.setPublisherConfirms(true);
final RabbitTemplate template = new RabbitTemplate(ccf);
final List<String> confirms = new ArrayList<String>();
final CountDownLatch latch = new CountDownLatch(2);
template.setConfirmCallback(new ConfirmCallback() {
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
if (ack) {
confirms.add(correlationData.getId());
latch.countDown();
}
}
@@ -628,8 +626,8 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
// 3.3.1 client
channel.basicConsume("foo", false, (Map) null, (Consumer) null);
verify(mockChannel).basicConsume("foo", false, (Map) null, (Consumer) null);
channel.basicConsume("foo", false, (Map) null, null);
verify(mockChannel).basicConsume("foo", false, (Map) null, null);
channel.basicQos(3, false);
verify(mockChannel).basicQos(3, false);
@@ -717,4 +715,63 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
assertThat(log.get(), containsString("NOT_FOUND - no exchange '" + exchange));
}
@Test
public void testConfirmReceivedAfterPublisherCallbackChannelScheduleClose() throws Exception {
final CountDownLatch latch = new CountDownLatch(40);
templateWithConfirmsEnabled.setConfirmCallback(new ConfirmCallback() {
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
latch.countDown();
}
});
ExecutorService executorService = Executors.newCachedThreadPool();
for (int i = 0; i < 20; i++) {
executorService.execute(new Runnable() {
@Override
public void run() {
templateWithConfirmsEnabled.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc"));
templateWithConfirmsEnabled.convertAndSend("BAD_ROUTE", (Object) "bad", new CorrelationData("cba"));
}
});
}
assertTrue(latch.await(10, TimeUnit.SECONDS));
assertNull(templateWithConfirmsEnabled.getUnconfirmed(0));
}
@Test
public void testReturnNotReceivedAfterPublisherCallbackChannelClose() throws Exception {
final CountDownLatch latch = new CountDownLatch(20);
templateWithReturnsEnabled.setMandatory(true);
templateWithReturnsEnabled.setReturnCallback(new ReturnCallback() {
@Override
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
latch.countDown();
}
});
ExecutorService executorService = Executors.newCachedThreadPool();
for (int i = 0; i < 20; i++) {
executorService.execute(new Runnable() {
@Override
public void run() {
templateWithReturnsEnabled.convertAndSend("BAD_ROUTE", (Object) "bad", new CorrelationData("cba"));
}
});
}
executorService.shutdown();
assertTrue(executorService.awaitTermination(10, TimeUnit.SECONDS));
Thread.sleep(100);
assertFalse(latch.getCount() == 0);
}
}

15
src/reference/docbook/amqp.xml
View File

@@ -563,12 +563,17 @@ public AmqpTemplate rabbitTemplate();
by calling <code>setConfirmCallback(ConfirmCallback callback)</code>. The callback
must implement this method:
</para>
<important>
Publisher Confirms only work when the channel is cached. Otherwise, the channel is closed after the
publish operation so, by definition, cannot receive the confirmation. Be sure to set the
<note>
When a rabbit template send operation completes, the channel is closed; this would preclude the reception
of confirms or returns in the case when the connection factory cache is full (when there is space in
the cache, the channel is not physically closed and the returns/confirms will proceed as normal).
When the cache is full, the framework defers the close for up to 5 seconds, in order to allow time
for the confirms/returns to be received. When using confirms, the channel will be closed when the
last confirm is received. When using only returns, the channel will remain open for the full
5 seconds. It is generally recommended to set the
connection factory's <code>channelCacheSize</code> to a large enough value so that the channel on which a
message is published is returned to the cache instead of being closed.
</important>
message is published is returned to the cache instead of being closed.
</note>
<programlisting language="java"><![CDATA[void confirm(CorrelationData correlationData, boolean ack, String cause);]]></programlisting>
<para>
The <classname>CorrelationData</classname> is an object supplied by the client when sending the