我正在使用以下的XML片段：
<int-amqp:inbound-channel-adapter acknowledge-mode="MANUAL" channel="commandQueue" concurrent-consumers="${commandConsumers:10}"
queue-names="commands" connection-factory="connectionFactory"/>
<int:channel id="commandQueue"/>
<int:channel id="commands"/>
<int:chain input-channel="commandQueue" output-channel="commands">
<int:delayer id="commandDelayer" default-delay="30000"/>
<int:json-to-object-transformer id="commandTransformer" type="com.airwatch.chat.command.Command"/>
</int:chain>
<int:payload-type-router input-channel="commands">
....
....
它正在执行以下任务：
- 从名为“commands”的RabbitMQ队列中消耗消息。
- 将消息执行延迟30秒。
- 在指定的延迟后继续执行消息。
我认为我知道为什么会发生这种情况。
Spring会重新安排存储在DelayHandler消息存储中的消息，一旦应用程序上下文完全初始化。请参考以下代码片段：
DelayHandler.java：public void onApplicationEvent(ContextRefreshedEvent event) {
if (!this.initialized.getAndSet(true)) {
this.reschedulePersistedMessages();
}
}
如果消息在应用程序启动之前已经存在于RabbitMQ队列中，在Spring上下文初始化期间，消息将从队列中获取并添加到DelayHandler的消息存储中。一旦完成上下文初始化，并且在此期间未释放消息，则上述代码段会重新安排相同的消息。
现在，当两个独立的线程执行同一条消息时，如果一个线程已经执行了该消息，则应该从消息存储中删除该消息，而另一个线程不应继续执行。
在线程执行时，DelayHandler.java中下面的代码片段允许第二个线程释放重复的消息，导致同一消息重复执行，因为消息存储是SimpleMessageStore的实例，并且没有进一步的检查来停止执行。
private void doReleaseMessage(Message<?> message) {
if (this.messageStore instanceof SimpleMessageStore
|| ((MessageStore) this.messageStore).removeMessage(message.getHeaders().getId()) != null) {
this.messageStore.removeMessageFromGroup(this.messageGroupId, message);
this.handleMessageInternal(message);
}
else {
if (logger.isDebugEnabled()) {
logger.debug("No message in the Message Store to release: " + message +
". Likely another instance has already released it.");
}
}
}
这是Spring Integration中的一个bug吗？