1.消息发送端(Producer)
public class OrderedProducer {
public static void main(String[] args) throws Exception {
//Instantiate with a producer group name.
MQProducer producer = new DefaultMQProducer("example_group_name");
//Launch the instance.
producer.start();
String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
for (int i = 0; i < 10; i++) {
int orderId = i % 10;
//Create a message instance, specifying topic, tag and message body.
Message msg = new Message("TopicTest", tags[i % tags.length], "KEY" + i,
("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
Integer id = (Integer) arg;
int index = id % mqs.size();
return mqs.get(index);
}
}, i);
System.out.printf("%s%n", sendResult);
}
//server shutdown
producer.shutdown();
}
}
1.1 MessageQueueSelector
MessageQueueSelector如其名,消息队列选择器, DefaultMQProducer 的send方法最后一个参数,会传入MessageQueueSelector的select回调方法中去
Topic创建后默认会有4个MessageQueue
要保证部分消息有序,需要发送端和消费端配合处理。 在发送端, 要做到把同一业务 ID的消息发送到同一个Message Queue; 在消费过程中,要做到从同一个 MessageQueue读取的消息不被并发处理, 这样才能达到部分有序。发送端使用MessageQueueSelector类来控制把消息发往哪个Message Queue
2.消息接收端(Consumer)
订阅MessageListenerOrderly消息
public class OrderedConsumer {
public static void main(String[] args) throws Exception {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("example_group_name");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.subscribe("TopicTest", "TagA || TagC || TagD");
consumer.registerMessageListener(new MessageListenerOrderly() {
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeOrderlyContext context) {
System.out.printf(Thread.currentThread().getName() + " Receive New Messages: " + msgs + "%n");
return ConsumeOrderlyStatus.SUCCESS;
}
});
consumer.start();
System.out.printf("Consumer Started.%n");
}
}
3.全局顺序消息
参考:
RocketMq基础认知
https://help.aliyun.com/document_detail/49323.html
网友评论