rocketMQ之局部有序消息

（1）生产者
//局部有序消息
public class Producer2 {
public static void main(String[] args) throws Exception{
Producer2 p = new Producer2();
p.sendMessage();
}

// 发送消息
void sendMessage() throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("producer_test");
// 指定NameServer地址
producer.setNamesrvAddr("192.168.191.133:9876"); // 多个ip以分号分隔
producer.start();
for (int i = 0; i < 20; i++) {
int orderId = i % 10;
// 自定义一个tag数组
            String[] tags = new String[]{"orders", "money", "goods"};
         // 构建消息
         Message msg = new Message("MyTopicOrder",
         tags[i % tags.length],
         "KEY"+i,
         ("hi " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
           // 发送10条消息到Topic为MyTopicOrder，tag为tags数组按顺序取值，
            // key值为“KEY”拼接上i的值，消息内容为“Hi”拼接上i的值
SendResult sendResult = producer.send(msg,
new MessageQueueSelector() {
public MessageQueue select(List<MessageQueue> mqs, Message m, Object arg1) {
// arg的值其实就是orderId
                        Integer id = (Integer) arg1;
                        // mqs是队列集合，也就是topic所对应的所有队列
                        int index = id % mqs.size();
                        // 这里根据前面的id对队列集合大小求余来返回所对应的队列
                        return mqs.get(index);
}
},
orderId);

System.out.printf("%s%n", sendResult);
}
producer.shutdown();
}
}

（2）消费者-有序消费
//消息有序消费:先启动消费者，再启动生产者
public class Consumer2 {

public static void main(String[] args) throws Exception{
Consumer2 c = new Consumer2();
c.receiveMessage();
}

void receiveMessage() throws Exception{
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_test");
// 指定NameServer地址，多个地址以 ; 隔开
consumer.setNamesrvAddr("192.168.191.133:9876");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
//设置consumer所订阅的Topic和Tag
        consumer.subscribe("MyTopicOrder", "orders || money || goods");
consumer.registerMessageListener(
new MessageListenerOrderly() {
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
for (MessageExt msg : msgs) {
String msgbody;
try {
msgbody = new String(msg.getBody(), "utf-8");
System.out.println("接收消息: " + msgbody);
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
//返回消费状态
                //SUCCESS 消费成功
                //SUSPEND_CURRENT_QUEUE_A_MOMENT 消费失败，暂停当前队列的消费
                return ConsumeOrderlyStatus.SUCCESS;
}
});
consumer.start();
System.out.printf("consumer start..%n");
}
}