rocketmq问题汇总-如何将特定消息发送至特定queue，消费者从特定queue消费

一时失言乱红尘 2023-06-19 04:16 7阅读 0赞

# rocketmq问题汇总-如何将特定消息发送至特定queue，消费者从特定queue消费 #

1.  业务描述
    
    > 由于业务需要这样一种场景，将消息按照id(业务id)尾号发送到对应的queue中，并启动10个消费者(单jvm，10个消费者组)，从对应的queue中集群消费，如下图1所示(假设有两个broker组成的集群)：   
    > ![图1][1]
2.  producer如何实现
    
    > producer只需发送消息时调用如下方法即可
    > 
    >     /**
    >      * 发送有序消息
    >      *
    >      * @param messageMap 消息数据
    >      * @param selector   队列选择器，发送时会回调
    >      * @param order      回调队列选择器时，此参数会传入队列选择方法,提供配需规则
    >      * @return 发送结果
    >      */
    >     public Result<SendResult> send(Message msg, MessageQueueSelector selector, Object arg)
    
    > 关键是如何实现MessageQueueSelector:
    > 
    >     class IDHashMessageQueueSelector implements MessageQueueSelector{
    >         public MessageQueue select(List<MessageQueue> mqs, Message msg,
    >                 Object arg) {
    >             int id = Integer.parseInt(arg.toString());
    >             int size = mqs.size();
    >             int index = id%size;
    >             return mqs.get(index);
    >         }
    >     }
    
    > 这样，所有的消息会根据消息的尾号，轮询的落到相应的queue上。参考图2，假设id=10001231，由于一共有20个queue，所以10001231%20=11，故消息会落到broker-b queue-1上。   
    > ![图2][2]
3.  consumer端如何实现
    
    > 针对consumer由于没有限制是顺序消费，故可以采用集群消费模式的DefaultMQPushConsumer，由于一个消费者消费一类queue，故需要10个consumer group，比如consumer group0需要消费的queue为broker-a queue-0和broker-b queue-0，如下图的概示：   
    > ![这里写图片描述][20160420114245543]   
    > 那么需要自己实现一个AllocateMessageQueueStrategy进行queue的分配，我们假设consumer group的名字格式需要提前定好，如xxx\{queueid\}ConsumerGroup，那么实现如下：
    > 
    >     public class AllocateMessageQueueByHashAveragely extends AllocateMessageQueueAveragely{
    >     private final Logger log = ClientLogger.getLog();
    >     @Override
    >     public String getName() {
    >         return super.getName()+"ByIDHash";
    >     }
    >     
    >     @Override
    >     public List<MessageQueue> allocate(String consumerGroup, String currentCID,
    >             List<MessageQueue> mqAll, List<String> cidAll) {
    >         //解析queue id
    >         char idChar = consumerGroup.charAt(consumerGroup.length() - "ConsumerGroup".length() - 1);
    >         int id = Integer.parseInt(idChar+"");
    >         List<MessageQueue> submq = new ArrayList<MessageQueue>();
    >         //根据queue id分配相应的MessageQueue
    >         for(MessageQueue mq : mqAll) {
    >             if(mq.getQueueId() == idChar || mq.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
    >                 submq.add(mq);
    >             }
    >         }
    >         if(submq.size() == 0) {
    >             log.warn("allocate err:"+consumerGroup+","+currentCID+","+cidAll+","+mqAll);
    >         }
    >         return super.allocate(consumerGroup, currentCID, submq, cidAll);
    >     }
    >     }
    
    > 借助AllocateMessageQueueAveragely来实现，以便有多个jvm的消费者时，能够进行集群消费，但是针对上面这个例子，消费者jvm实例不能超过2个，至于为什么，参照下图：   
    > ![这里写图片描述][20160420115405604]
4.  over

[1]: https://img-blog.csdn.net/20160420120139821
[2]: https://img-blog.csdn.net/20160420120200255
[20160420114245543]: https://img-blog.csdn.net/20160420114245543
[20160420115405604]: https://img-blog.csdn.net/20160420115405604