我正在审核一个用Java编写的客户端-服务器应用程序。服务器接收JMS消息并处理它们,但是这些消息可能以意外的顺序到达,而取消消息可能在订单消息之前到达。您如何处理这种情况?您是在MDB中处理吗?
针对这种情况,有哪些策略或模式可供选择?
否则,可以看一下message store模式。无论如何,值得查看EAI网站。
如果系统能够处理乱序消息,它将变得更加灵活。我过去用来解决这个问题的模式是使用延迟队列(在金融界每天处理800万条消息的系统中)。
在您的示例中,如果我收到了一个我尚未接收到的订单删除请求,我会将其延迟一段时间并重试。如果我仍然不知道被要求删除的订单,我会引发某种错误(回复原始发送者,向特殊错误队列发送消息等)。
至于延迟队列的实现,这可以是另一个JMS队列,具有可以接受要延迟的消息的服务。然后定期读取延迟的消息,并检查是否已经过期并重新提交消息到原始目标队列。
String message;
String messageKey;
ReentrantLock messageKeyLock;
partitioningSupport.getFairLock().lock();
try {
// use DUPS_OK_ACKNOWLEDGE with deduplication service which improve performance of sequential read
message = (String) jmsTemplate.receiveAndConvert(QUEUE);
if (message == null || deduplicationService.deduplicate(md5(message)))
continue;
messageKey = findByXPath(path, message)
messageKeyLock = partitioningSupport.getPartitionLock(messageKey);
} finally {
partitioningSupport.getFairLock().unlock();
}
messageKeyLock.lock();
try {
// parallel message processing
} finally {
messageKeyLock.unlock();
}
Implementation
import static org.apache.commons.lang3.RandomUtils.nextInt;
import static org.apache.commons.lang3.StringUtils.isBlank;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;
public class PartitioningSupport {
private final ConcurrentHashMap<Integer, ReentrantLock> locks = new ConcurrentHashMap<>();
private final ReentrantLock fairLock = new ReentrantLock(true);
private final int diversity;
public PartitioningSupport() {
this(0xff);
}
public PartitioningSupport(int diversity) {
this.diversity = diversity;
}
public ReentrantLock getPartitionLock(String messageKey) {
fairLock.lock();
try {
int partition = partition(messageKey);
ReentrantLock lock = locks.get(partition);
if (lock == null) {
lock = new ReentrantLock(true);
locks.put(partition, lock);
}
return lock;
} finally {
fairLock.unlock();
}
}
private int partition(String key) {
return (isBlank(key) ? nextInt() : key.hashCode()) & diversity;
}
public ReentrantLock getFairLock() {
return fairLock;
}
}
Test
import static java.lang.Integer.parseInt;
import static java.lang.String.format;
import static java.lang.System.out;
import static java.lang.Thread.sleep;
import static java.util.concurrent.Executors.newFixedThreadPool;
import static org.apache.commons.lang3.RandomUtils.nextInt;
import static org.apache.commons.lang3.RandomUtils.nextLong;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;
import org.junit.jupiter.api.Test;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
public class PartitioningSupportTest {
private BlockingQueue<String> queue = new LinkedBlockingDeque<>();
private List<Future<?>> results = new ArrayList<>();
private ExecutorService consumers = newFixedThreadPool(10, new ThreadFactoryBuilder().setNameFormat("consumer-%s").build());
private PartitioningSupport partitioningSupport = new PartitioningSupport();
private volatile ConcurrentHashMap<String, AtomicInteger> ids;
private int repeatTest = 10;
private int uniqueKeysCount = 1; // 100
private int totalMessagesCount = 1000;
@Test
public void testProcessingOrder() throws InterruptedException, ExecutionException {
for (int testIter = 0; testIter < repeatTest; testIter++) {
ids = new ConcurrentHashMap<>();
results = new ArrayList<>();
for (int messageIter = 1; messageIter <= totalMessagesCount; messageIter++) {
String messageKey = "message-" + nextInt(0, uniqueKeysCount);
ids.putIfAbsent(messageKey, new AtomicInteger());
queue.put(format("%s.%s", messageKey, messageIter));
}
for (int i = 0; i < totalMessagesCount; i++)
results.add(consumers.submit(this::consume));
for (Future<?> result : results)
result.get();
}
consumers.shutdown();
}
private void consume() {
try {
String message;
String messageKey;
ReentrantLock messageKeyLock;
partitioningSupport.getFairLock().lock();
try {
message = queue.take();
messageKey = message.substring(0, message.indexOf('.'));
messageKeyLock = partitioningSupport.getPartitionLock(messageKey);
} finally {
partitioningSupport.getFairLock().unlock();
}
messageKeyLock.lock();
try {
sleep(nextLong(1, 10));
int ordinal = parseInt(message.substring(message.indexOf('.') + 1));
int previous = ids.get(messageKey).getAndSet(ordinal);
out.printf("processed: %s - %s%n", messageKey, ordinal);
assertTrue(ordinal > previous, format("broken order %s [%s -> %s]", messageKey, previous, ordinal));
} finally {
messageKeyLock.unlock();
}
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}