如何处理 JMS 中的消息顺序？

Question

我正在审查一个用 Java 编写的客户端-服务器应用程序。服务器接收 JMS 消息并对其进行处理，但消息可能以意外的顺序出现，并且取消可能在订单消息之前到达。你如何处理这样的情况？你是在mdb里做的吗？

这种场景有哪些策略或模式？

Answer 1

到目前为止，我知道，这被称为“无序”交付，是 JMS 系统的服务质量 (QoS) 属性的一部分。我不认为它是 JMS 规范的一部分，但一些提供商可能支持它。这取决于您使用的特定 JMS 实现。

但是请注意，JMS 旨在以一种分配负载的方式将消息分发给几个消费者。如果消息必须以有序的方式传递，这是不可能的——它基本上会导致消息传递的序列化，并且消息无法同时处理。

wikipedia 说得比我好：

JMS 队列 一个暂存区，包含已发送和正在发送的消息 等待阅读。注意， 与队列名称相反 建议，消息不一定是 按照发送的顺序交付。如果 消息驱动的 bean 池包含更多 不止一个实例，那么消息可以是 同时处理，因此它是 可能是稍后的消息 比之前的处理更快。 JMS 队列只保证每个 消息只处理一次。

那么，使用 JMS 来实现带外取消请求并不容易。两个想法：

• 在数据库中存储与每条消息相对应的票证，可用于轻松取消消息。当消息传递时，MDB 检查相应的票证是否仍然有效。如果是，则继续，如果不是，则丢弃该消息。
• 尝试将 MDB 池大小设置为 1。也许在这种情况下，将订购交货。更改池大小是应用程序。特定于服务器，但大多数都支持每个 bean 池大小。

否则，不妨看看message store 模式。无论如何，值得查看EAI 网站。

Answer 2

如果您的系统能够处理乱序消息，您的系统将会更加灵活。我过去用来解决这个问题的模式是使用延迟队列（在金融界每天处理 800 万条消息的系统上）。

在您的示例中，如果我收到一个我尚未收到的订单的删除，我会延迟一段时间并重试。如果我仍然对被要求删除的订单一无所知，我会提出某种错误（回复原始发件人，向特殊错误队列发送消息，...）。

关于延迟队列的实现，这可以是另一个JMS 队列，其服务可以接受要延迟的消息。然后它会定期读取延迟的消息，并检查延迟的时间段是否已过期，然后将消息重新提交到原始目标队列。

Answer 3

我赞同关于检查 EAI 网站及其所依据的书的建议（关于 MOM 和 MOM 模式的精彩文本）。

不过，我个人会调查Resequencer。

Answer 4

How to assure the sequence of message received by mdb? 是关于服务器端的一个类似主题，有人指出ActiveMQ 可能有一个保留顺序的解决方案。我想这使它变得更加具体。

Answer 5

JMS 队列一般应被视为FIFO 队列。
根据IBM MQ documentation的说法，订购被宠坏的原因是：

• 多个目的地
• 多个生产者
• 多个消费者
• 发布和订阅（意味着订阅的多个实例）

ActiveMQ 的类似语句

ActiveMQ 将保留单个生产者发送给主题上所有消费者的消息的顺序。如果队列中有单个消费者，则单个生产者发送的消息顺序也将被保留。 如果您在单个队列上有多个消费者，消费者将竞争消息，ActiveMQ 将在它们之间进行负载平衡，因此将丢失顺序。

您需要由同一线程（按顺序）处理同一组的消息，而不是重新排序它们。 Kafka 为您提供基于message key 的消息智能分区。 ActiveMQ 有message groups 的概念，它利用了消息头。

如果您不能使用上述方法，请考虑在消费者应用程序中使用 java 公平锁的分区示例。从队列中读取消息和派生分区应该是同步的，实际处理可以并行。

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();
}

具有 10 个密钥多样性（唯一密钥的数量）、10 个消费者线程和 255 个分区，锁定是显着的

在 1000 个密钥的多样性和其他相同的情况下，锁定相当随意且不显着（等待的概率相对较小）。

实施

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;
    }
}

测试

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);
        }
    }
}