Вот окончательное решение, которое я реализовал. На «интервал повтора» приходится 1 очередь, на каждую очередь повтора - 1 обмен. Все они передаются настраиваемому RepublishRecoverer, который создает список восстановителей.
К сообщению добавляется настраиваемый заголовок RetryCount, и в зависимости от значения RetryCount сообщение публикуется в правильном обмене / очереди с другим «сроком действия». Каждая очередь повторных попыток настраивается с помощью DLX, для которого установлено значение «regular_exchange» (т.е. запросы поступают в обычную очередь).
<rabbit:template id="genericTemplateWithRetry" connection-factory="connectionFactory" exchange="regular_exchange" retry-template="retryTemplate"/>
<!-- Create as many templates as retryAttempts (1st arg) in customRetryTemplate-->
<rabbit:template id="genericRetryTemplate1" connection-factory="consumerConnFactory" exchange="retry_exchange_1"/>
<rabbit:template id="genericRetryTemplate2" connection-factory="consumerConnFactory" exchange="retry_exchange_2"/>
<rabbit:template id="genericRetryTemplate3" connection-factory="consumerConnFactory" exchange="retry_exchange_3"/>
<rabbit:template id="genericRetryTemplate4" connection-factory="consumerConnFactory" exchange="retry_exchange_4"/>
<rabbit:template id="genericRetryTemplate5" connection-factory="consumerConnFactory" exchange="retry_exchange_5"/>
<rabbit:queue name="regular_requests_queue"/>
<!-- Create as many queues as retryAttempts (1st arg) in customRetryTemplate -->
<rabbit:queue name="retry_requests_queue_1">
<rabbit:queue-arguments>
<entry key="x-dead-letter-exchange" value="regular_exchange" />
</rabbit:queue-arguments>
</rabbit:queue>
<rabbit:queue name="retry_requests_queue_2">
<rabbit:queue-arguments>
<entry key="x-dead-letter-exchange" value="regular_exchange" />
</rabbit:queue-arguments>
</rabbit:queue>
<rabbit:queue name="retry_requests_queue_3">
<rabbit:queue-arguments>
<entry key="x-dead-letter-exchange" value="regular_exchange" />
</rabbit:queue-arguments>
</rabbit:queue>
<rabbit:queue name="retry_requests_queue_4">
<rabbit:queue-arguments>
<entry key="x-dead-letter-exchange" value="regular_exchange" />
</rabbit:queue-arguments>
</rabbit:queue>
<rabbit:queue name="retry_requests_queue_5">
<rabbit:queue-arguments>
<entry key="x-dead-letter-exchange" value="regular_exchange" />
</rabbit:queue-arguments>
</rabbit:queue>
<rabbit:direct-exchange name="regular_exchange">
<rabbit:bindings>
<rabbit:binding queue="regular_requests_queue" key="v1-regular-request"/>
</rabbit:bindings>
</rabbit:direct-exchange>
<!-- Create as many exchanges as retryAttempts (1st arg) in customRetryTemplate -->
<rabbit:direct-exchange name="retry_exchange_1">
<rabbit:bindings>
<rabbit:binding queue="retry_requests_queue_1" key="v1-regular-request"/>
</rabbit:bindings>
</rabbit:direct-exchange>
<rabbit:direct-exchange name="retry_exchange_2">
<rabbit:bindings>
<rabbit:binding queue="retry_requests_queue_2" key="v1-regular-request"/>
</rabbit:bindings>
</rabbit:direct-exchange>
<rabbit:direct-exchange name="retry_exchange_3">
<rabbit:bindings>
<rabbit:binding queue="retry_requests_queue_3" key="v1-regular-request"/>
</rabbit:bindings>
</rabbit:direct-exchange>
<rabbit:direct-exchange name="retry_exchange_4">
<rabbit:bindings>
<rabbit:binding queue="retry_requests_queue_4" key="v1-regular-request"/>
</rabbit:bindings>
</rabbit:direct-exchange>
<rabbit:direct-exchange name="retry_exchange_5">
<rabbit:bindings>
<rabbit:binding queue="retry_requests_queue_5" key="v1-regular-request"/>
</rabbit:bindings>
</rabbit:direct-exchange>
<!-- retry config begin -->
<!-- Pass in all templates and exchanges created as list/array arguments below -->
<bean id="customRetryRecoverer" class="com.test.listeners.CustomRetryRecoverer">
<!-- Pass in list of templates -->
<constructor-arg>
<list>
<ref bean="genericRetryTemplate1"/>
<ref bean="genericRetryTemplate2"/>
<ref bean="genericRetryTemplate3"/>
<ref bean="genericRetryTemplate4"/>
<ref bean="genericRetryTemplate5"/>
</list>
</constructor-arg>
<!-- Pass in array of exchanges -->
<constructor-arg value="retry_exchange_1,retry_exchange_2,retry_exchange_3,retry_exchange_4,retry_exchange_5"/>
<constructor-arg ref="customRetryTemplate"/>
</bean>
<bean id="retryInterceptor"
class="org.springframework.amqp.rabbit.config.StatefulRetryOperationsInterceptorFactoryBean">
<property name="messageRecoverer" ref="customRetryRecoverer"/>
<property name="retryOperations" ref="retryTemplate"/>
<property name="messageKeyGenerator" ref="msgKeyGenerator"/>
</bean>
<bean id="retryTemplate" class="org.springframework.retry.support.RetryTemplate">
<property name="retryPolicy">
<bean class="org.springframework.retry.policy.SimpleRetryPolicy">
<!-- Set to 1 - just for the initial attempt -->
<property name="maxAttempts" value="1"/>
</bean>
</property>
</bean>
<bean id="customRetryTemplate" class="com.test.retry.CustomRetryTemplate">
<constructor-arg value="5"/> <!-- max attempts -->
<constructor-arg value="3000"/> <!-- Initial interval -->
<constructor-arg value="5"/> <!-- multiplier for backoff -->
</bean>
<!-- retry config end -->
Вот код CustomRetryRecoverer:
public class CustomRetryRecoverer extends
RepublishMessageRecoverer {
private static final String RETRY_COUNT_HEADER_NAME = "RetryCount";
private List<RepublishMessageRecoverer> retryExecutors = new ArrayList<RepublishMessageRecoverer>();
private TriggersRetryTemplate retryTemplate;
public TriggersRetryRecoverer(AmqpTemplate[] retryTemplates, String[] exchangeNames, TriggersRetryTemplate retryTemplate) {
super(retryTemplates[0], exchangeNames[0]);
this.retryTemplate = retryTemplate;
//Get lower of the two array sizes
int executorCount = (exchangeNames.length < retryTemplates.length) ? exchangeNames.length : retryTemplates.length;
for(int i=0; i<executorCount; i++) {
createRetryExecutor(retryTemplates[i], exchangeNames[i]);
}
//If not enough exchanges/templates provided, reuse the last exchange/template for the remaining retry recoverers
if(retryTemplate.getMaxRetryCount() > executorCount) {
for(int i=executorCount; i<retryTemplate.getMaxRetryCount(); i++) {
createRetryExecutor(retryTemplates[executorCount-1], exchangeNames[executorCount-1]);
}
}
}
@Override
public void recover(Message message, Throwable cause) {
if(getRetryCount(message) < retryTemplate.getMaxRetryCount()) {
incrementRetryCount(message);
//Set the expiration of the retry message
message.getMessageProperties().setExpiration(String.valueOf(retryTemplate.getNextRetryInterval(getRetryCount(message)).longValue()));
RepublishMessageRecoverer retryRecoverer = null;
if(getRetryCount(message) != null && getRetryCount(message) > 0) {
retryRecoverer = retryExecutors.get(getRetryCount(message)-1);
} else {
retryRecoverer = retryExecutors.get(0);
}
retryRecoverer.recover(message, cause);
} else {
//Retries exchausted - do nothing
}
}
private void createRetryExecutor(AmqpTemplate template, String exchangeName) {
RepublishMessageRecoverer retryRecoverer = new RepublishMessageRecoverer(template, exchangeName);
retryRecoverer.errorRoutingKeyPrefix(""); //Set KeyPrefix to "" so original key is reused during retries
retryExecutors.add(retryRecoverer);
}
private Integer getRetryCount(Message msg) {
Integer retryCount;
if(msg.getMessageProperties().getHeaders().get(RETRY_COUNT_HEADER_NAME) == null) {
retryCount = 1;
} else {
retryCount = (Integer) msg.getMessageProperties().getHeaders().get(RETRY_COUNT_HEADER_NAME);
}
return retryCount;
}
private void incrementRetryCount(Message msg) {
Integer retryCount;
if(msg.getMessageProperties().getHeaders().get(RETRY_COUNT_HEADER_NAME) == null) {
retryCount = 1;
} else {
retryCount = (Integer) msg.getMessageProperties().getHeaders().get(RETRY_COUNT_HEADER_NAME)+1;
}
msg.getMessageProperties().getHeaders().put(RETRY_COUNT_HEADER_NAME, retryCount);
}
}
Код для CustomRetryTemplate здесь не размещен, но он содержит простые переменные для maxRetryCount, initialInterval и multiplier.
person
alam86
schedule
20.10.2015
