51.RocketMQ 顺序消费

Posted 2020-09-11 司广孟

3种不同模式的Producer

• NormalProducer（普通）
• OrderProducer（顺序）
• TransactionProducer（事务）

 

生产者

 1 /**
 2  * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 3  *
 4  * Licensed under the Apache License, Version 2.0 (the "License");
 5  * you may not use this file except in compliance with the License.
 6  * You may obtain a copy of the License at
 7  *
 8  *     http://www.apache.org/licenses/LICENSE-2.0
 9  *
10  * Unless required by applicable law or agreed to in writing, software
11  * distributed under the License is distributed on an "AS IS" BASIS,
12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13  * See the License for the specific language governing permissions and
14  * limitations under the License.
15  */
16 package com.alibaba.rocketmq.example.ordermessage;
17 
18 import java.util.List;
19 
20 import com.alibaba.rocketmq.client.exception.MQBrokerException;
21 import com.alibaba.rocketmq.client.exception.MQClientException;
22 import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
23 import com.alibaba.rocketmq.client.producer.MQProducer;
24 import com.alibaba.rocketmq.client.producer.MessageQueueSelector;
25 import com.alibaba.rocketmq.client.producer.SendResult;
26 import com.alibaba.rocketmq.common.message.Message;
27 import com.alibaba.rocketmq.common.message.MessageQueue;
28 import com.alibaba.rocketmq.remoting.exception.RemotingException;
29 
30 
31 /**
32  * Producer，发送顺序消息
33  */
34 public class Producer {
35     public static void main(String[] args) {
36         try {
37             MQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
38 
39             producer.start();
40 
41             String[] tags = new String[] { "TagA", "TagB", "TagC", "TagD", "TagE" };
42 
43             for (int i = 0; i < 100; i++) {
44                 // 订单ID相同的消息要有序
45                 int orderId = i % 10;
46                 Message msg = new Message("TopicTestjjj", tags[i % tags.length], "KEY" + i,
47                     ("Hello RocketMQ " + i).getBytes());
48 
49                 SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
50                     @Override
51                     public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
52                         Integer id = (Integer) arg;
53                         int index = id % mqs.size();
54                         return mqs.get(index);
55                     }
56                 }, orderId);
57 
58                 System.out.println(sendResult);
59             }
60 
61             producer.shutdown();
62         }
63         catch (MQClientException e) {
64             e.printStackTrace();
65         }
66         catch (RemotingException e) {
67             e.printStackTrace();
68         }
69         catch (MQBrokerException e) {
70             e.printStackTrace();
71         }
72         catch (InterruptedException e) {
73             e.printStackTrace();
74         }
75     }
76 }

消费者，只允许有一个线程在接收消息，因为broker只保证是顺序到达，不保证在消费的时候也能顺序消费。

 1 /**
 2  * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 3  *
 4  * Licensed under the Apache License, Version 2.0 (the "License");
 5  * you may not use this file except in compliance with the License.
 6  * You may obtain a copy of the License at
 7  *
 8  *     http://www.apache.org/licenses/LICENSE-2.0
 9  *
10  * Unless required by applicable law or agreed to in writing, software
11  * distributed under the License is distributed on an "AS IS" BASIS,
12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13  * See the License for the specific language governing permissions and
14  * limitations under the License.
15  */
16 package com.alibaba.rocketmq.example.ordermessage;
17 
18 import java.util.List;
19 import java.util.concurrent.atomic.AtomicLong;
20 
21 import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
22 import com.alibaba.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
23 import com.alibaba.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
24 import com.alibaba.rocketmq.client.consumer.listener.MessageListenerOrderly;
25 import com.alibaba.rocketmq.client.exception.MQClientException;
26 import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
27 import com.alibaba.rocketmq.common.message.MessageExt;
28 
29 
30 /**
31  * 顺序消息消费，带事务方式（应用可控制Offset什么时候提交）
32  */
33 public class Consumer {
34 
35     public static void main(String[] args) throws MQClientException {
36         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_3");
37         /**
38          * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
39          * 如果非第一次启动，那么按照上次消费的位置继续消费
40          */
41         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
42 
43         consumer.subscribe("TopicTest", "TagA || TagC || TagD");
44 
45         consumer.registerMessageListener(new MessageListenerOrderly() {
46             AtomicLong consumeTimes = new AtomicLong(0);
47 
48 
49             @Override
50             public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
51                 context.setAutoCommit(false);
52                 System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);
53                 this.consumeTimes.incrementAndGet();
54                 if ((this.consumeTimes.get() % 2) == 0) {
55                     return ConsumeOrderlyStatus.SUCCESS;
56                 }
57                 else if ((this.consumeTimes.get() % 3) == 0) {
58                     return ConsumeOrderlyStatus.ROLLBACK;
59                 }
60                 else if ((this.consumeTimes.get() % 4) == 0) {
61                     return ConsumeOrderlyStatus.COMMIT;
62                 }
63                 else if ((this.consumeTimes.get() % 5) == 0) {
64                     context.setSuspendCurrentQueueTimeMillis(3000);
65                     return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
66                 }
67 
68                 return ConsumeOrderlyStatus.SUCCESS;
69             }
70         });
71 
72         consumer.start();
73 
74         System.out.println("Consumer Started.");
75     }
76 
77 }