activemq是如何实现消息分组的

                if (!fullConsumers.contains(s)) {
                     if (!s.isFull()) {
                         if (dispatchSelector.canSelect(s, node) && assignMessageGroup(s, (QueueMessageReference)node) && !((QueueMessageReference) node).isAcked() ) {
                             // Dispatch it.
                             s.add(node);
                             target = s;
                             break;
                         }
                     } else {
                         // no further dispatch of list to a full consumer to
                         // avoid out of order message receipt
                         fullConsumers.add(s);
                         LOG.trace("Subscription full {}", s);
                     }
                 }
    protected boolean assignMessageGroup(Subscription subscription, QueueMessageReference node) throws Exception {
         boolean result = true;
         // Keep message groups together.
         String groupId = node.getGroupID();
         int sequence = node.getGroupSequence();
         if (groupId != null) {
            MessageGroupMap messageGroupOwners = getMessageGroupOwners();
             // If we can own the first, then no-one else should own the
             // rest.
             if (sequence == 1) {
 assignGroup(subscription, messageGroupOwners, node, groupId);
             } else {
                // Make sure that the previous owner is still valid, we may
                 // need to become the new owner.
                 ConsumerId groupOwner;
                groupOwner = messageGroupOwners.get(groupId);
                 if (groupOwner == null) {
 assignGroup(subscription, messageGroupOwners, node, groupId);
                 } else {
                     if (groupOwner.equals(subscription.getConsumerInfo().getConsumerId())) {
                         // A group sequence < 1 is an end of group signal.
                         if (sequence < 0) {
                             messageGroupOwners.removeGroup(groupId);
                             subscription.getConsumerInfo().setLastDeliveredSequenceId(subscription.getConsumerInfo().getLastDeliveredSequenceId() - 1);
                         }
                     } else {
                         result = false;
                     }
                 }
             }
         }
        return result;
    }