五分钟带你玩转rocketMQ（六）队列难题——如何顺序消费

1. package cn.baocl.rocketmq.processor;
2. 
3. 
4. import com.alibaba.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
5. import com.alibaba.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
6. import com.alibaba.rocketmq.client.consumer.listener.MessageListenerOrderly;
7. import com.alibaba.rocketmq.common.message.MessageExt;
8. import org.slf4j.Logger;
9. import org.slf4j.LoggerFactory;
10. import org.springframework.stereotype.Component;
11. 
12. import java.util.List;
13. import java.util.concurrent.atomic.AtomicLong;
14. 
15. @Component
16. public class MQConsumeMsgListenerProcessor implements MessageListenerOrderly {
17. private static final Logger logger = LoggerFactory.getLogger(MQConsumeMsgListenerProcessor.class);
18. 
19. @Override
20. public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext context) {
21. //原子类 可以使消费者同步进行
22.         AtomicLong consumeTimes = new AtomicLong(0);
23.         context.setAutoCommit(false);
24.         consumeTimes.incrementAndGet();
25. if ((consumeTimes.get() % 2) == 0) {
26. return ConsumeOrderlyStatus.SUCCESS;
27.         } else if ((consumeTimes.get() % 3) == 0) {
28. return ConsumeOrderlyStatus.ROLLBACK;
29.         } else if ((consumeTimes.get() % 4) == 0) {
30. return ConsumeOrderlyStatus.COMMIT;
31.         } else if ((consumeTimes.get() % 5) == 0) {
32.             context.setSuspendCurrentQueueTimeMillis(3000);
33. return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
34.         }
35. return ConsumeOrderlyStatus.SUCCESS;
36.     }
37. 
38. 
39. }

1. package cn.baocl.rocketmq.controllor;
2. 
3. import cn.baocl.rocketmq.entity.TestVo;
4. import com.alibaba.rocketmq.client.exception.MQBrokerException;
5. import com.alibaba.rocketmq.client.exception.MQClientException;
6. import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
7. import com.alibaba.rocketmq.client.producer.MessageQueueSelector;
8. import com.alibaba.rocketmq.client.producer.SendCallback;
9. import com.alibaba.rocketmq.client.producer.SendResult;
10. import com.alibaba.rocketmq.common.message.Message;
11. import com.alibaba.rocketmq.common.message.MessageQueue;
12. import com.alibaba.rocketmq.remoting.common.RemotingHelper;
13. import com.alibaba.rocketmq.remoting.exception.RemotingException;
14. import org.slf4j.Logger;
15. import org.slf4j.LoggerFactory;
16. import org.springframework.beans.factory.annotation.Autowired;
17. import org.springframework.web.bind.annotation.RequestMapping;
18. import org.springframework.web.bind.annotation.RestController;
19. 
20. import javax.annotation.Resource;
21. import java.util.List;
22. 
23. 
24. @RestController
25. @RequestMapping("/test")
26. public class TestControllor {
27. private static final Logger logger = LoggerFactory.getLogger(TestControllor.class);
28. 
29. /**
30.      * 使用RocketMq的生产者
31.      */
32. @Resource(name = "customRocketMQProducer")
33. private DefaultMQProducer producer;
34. 
35. @RequestMapping("/send")
36. public void send() throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
37. //定义tags
38.         String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
39. for (int i = 0; i < 100; i++) {
40. int orderId = i % 10;
41. //指定每条消息发送到某个tags
42. Message msg = new Message("DemoTopic", tags[i % tags.length], "KEY" + i,
43.                     ("现在排号到:" + i).getBytes());
44. //顺序发送方法
45. SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
46. @Override
47. public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
48. Integer id = (Integer) arg;
49. int index = id % mqs.size();
50. return mqs.get(index);
51.                 }
52.             }, orderId);
53. //orderId 就是队列编号 ，默认有4个队列
54.             System.out.print("此条消息id为:"+i);
55.             System.out.printf("%s%n", sendResult);
56.         }
57.     }
58. }