顺序消费实战
顺序消费分为两种：
- 全局有序：适用于并发度不大，并且对消息要求严格一致性的场景下
通过创建一个 topic，并且该 topic 下只有一个队列，那么生产者向着一个队列中发消息，消费者也在这一个队列中消费消息，来保证消息的有序性 - 局部有序：适用于对性能要求比较高的场景，在设计层面将需要保证有序的消息放在 Topic 下的同一个队列即可保证有序
全局有序
要保证全局有序的话，我们先通过上边启动的 Dashboard 项目，创建一个只有一个队列的 Topic
将 写队列和读队列 都设置为 1 个，perm 设置为6（perm，2：只写； 4-只读； 6-读写；）
全局有序流程图如下：
首先消费者主启动类如下：
@SpringBootApplication
@EnableBinding({CustomSink.class })
public class OrderlyConsumerApplication {
@Value("${server.port}")
private int port;
public static void main(String[] args) {
SpringApplication.run(OrderlyConsumerApplication.class, args);
System.out.println("【【【【【 OrderlyConsumerApplication 启动成功！！！ 】】】】】");
}
// 定义两个通道，input 接收全局有序消息，input2 接收局部有序消息
@StreamListener("input")
public void receiveInput(String receiveMsg) {
System.out.println(port + " port, input receive: " + receiveMsg);
}
@StreamListener("input2")
public void receiveInput2(String receiveMsg) {
System.out.println(port + " port, input2 receive: " + receiveMsg);
}
}自定义 CustomSink 如下：
public interface CustomSink extends Sink {
/**
* Input channel name.
*/
String INPUT2 = "input2";
/**
* @return input channel.
*/
@Input(CustomSink.INPUT2)
SubscribableChannel input2();
}配置类 application.properties 如下：
=mq_orderly_consumer
server.port=9530
# configure the nameserver of rocketmq
-server=127.0.0.1:9876
spring.cloud.stream.rocketmq.binder.group=mq_producer_group
# configure the input binding named input
spring.cloud.stream.bindings.input.destination=Global-Orderly-Topic
spring.cloud.stream.bindings.input.content-type=application/json
spring.cloud.stream.bindings.input.group=Global-Orderly-Topic-group
spring.cloud.stream.rocketmq.bindings.input.consumer.orderly=true
# configure the input binding named input
spring.cloud.stream.bindings.input2.destination=Partly-Orderly-Topic
spring.cloud.stream.bindings.input2.content-type=application/json
spring.cloud.stream.bindings.input2.group=Partly-Orderly-Topic-group
spring.cloud.stream.rocketmq.bindings.input2.consumer.orderly=true全局有序生产者代码如下：
public class GlobalProducer {
public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer(
"producer_group",
true);
producer.setNamesrvAddr("218.95.37.160:9876");
producer.start();
for (int i = 0; i < 12; i++) {
Message msg = new Message(
"Global-Orderly-Topic",
"Global_Orderly_Tag",
("( " + i + " )message from GlobalProducer").getBytes());
msg.setKeys("Global_Orderly_Tag");
producer.send(msg);
}
System.out.println("Send Finished.");
}
}先启动消费者，再启动生产者，即可看到在消费者端，消息被有序消费
局部有序
局部有序的话，我们将需要保证有序的消息放在同一个 Topic 下的队列即可保证有序，这里设计的让 OrderId 相同的消息放在同一个队列中发送，流程图如下：
在局部有序中，消费者依然使用全局有序中的消费者，局部生产者代码如下：
public class PartlyProducer {
public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer(
"producer_group",
true);
producer.setNamesrvAddr("127.0.0.1:9876");
producer.start();
/**
* orderId = 1 的消息，需要按照 step 的顺序进行消费
* orderId = 2 的消息，需要按照 step 的顺序进行消费
*/
List<Order> list = new ArrayList<>();
for (int i = 1; i <= 3; i ++) {
Order order = new Order();
order.orderId = 1;
order.step = i;
list.add(order);
}
for (int i = 5; i <= 8; i ++) {
Order order = new Order();
order.orderId = 2;
order.step = i;
list.add(order);
}
System.out.println(list);
int size = list.size();
for (int i = 0; i < size; i++) {
Order order = list.get(i);
Message msg = new Message(
"Partly-Orderly-Topic",
"Partly_Orderly_Tag",
(order.toString()).getBytes());
msg.setKeys("Partly_Orderly_Tag");
/**
* 这里发送消息的时候，根据 orderId 来选择对应发送的队列
*/
producer.send(msg, new MessageQueueSelector() {
@Override
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
int orderId = (int)arg;
int idx = orderId % mqs.size();
return mqs.get(idx);
}
}, order.orderId);
}
System.out.println("Send Finished.");
}
public static class Order {
int orderId;
int step;
@Override
public String toString() {
return "Order{" +
"orderId=" + orderId +
", step=" + step +
'}';
}
}
}
