alibaba rocket mq的串行及并行处理

项目中用到rocket mq的方式有多种，

第一种，严格按照时间消费的模式，这种模式需要用串行方式，生产者生产的时候，这时候生产者需要往特定的队列里有序push：

                     SendResult result = producer.send(msg, new MessageQueueSelector(){
@Override
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
Integer id = arg.hashCode();
int index = id % mqs.size();
return mqs.get(index);
}

}, dataAsyncEvent.getDataType());

 

                   消费者也要按照顺序严格有序消费，用这个有序的监听者：

                   //同一队列的消息同一时刻只能一个线程消费，可保证消息在同一队列严格有序消费

 

 consumer.registerMessageListener(new MessageListenerOrderly() {
@Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
                    ConsumeOrderlyContext context) {

                    return ConsumeOrderlyStatus.CONSUME_SUCCESS;

             }

                   public interface MessageListenerOrderly extends MessageListener {
    /**
     * 方法抛出异常等同于返回 ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT<br>
     * P.S: 建议应用不要抛出异常
     * 
     * @param msgs
     *            msgs.size() >= 1<br>
     *            DefaultMQPushConsumer.consumeMessageBatchMaxSize=1，默认消息数为1
     * @param context
     * @return
     */
    public ConsumeOrderlyStatus consumeMessage(final List<MessageExt> msgs,
            final ConsumeOrderlyContext context);
} 

 

第二种：不追求时间顺序，只要把生产出来的事件全部消费完就可以。这种可以用并行的方式处理，效率高很多：

               生产者：

               SendResult result = producer.send(msg);

                消费者：（用此接口处理）

               consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                    ConsumeConcurrentlyContext context) {

                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

             }

                public interface MessageListenerConcurrently extends MessageListener {
    /**
     * 方法抛出异常等同于返回 ConsumeConcurrentlyStatus.RECONSUME_LATER<br>
     * P.S: 建议应用不要抛出异常
     * 
     * @param msgs
     *            msgs.size() >= 1<br>
     *            DefaultMQPushConsumer.consumeMessageBatchMaxSize=1，默认消息数为1
     * @param context
     * @return
     */
    public ConsumeConcurrentlyStatus consumeMessage(final List<MessageExt> msgs,
            final ConsumeConcurrentlyContext context);
}