RocketMQ-消息消费模式 顺序消费
RocketMQ-消息消费模式
集群模式
在消费模式为集群的情况下,如果机器是集群的,消息只会给集群中的其中一台机器消费到
集群模式的演示(本身就默认)
假设我们生产者生产了十条信息 ，当我们集群了两台消费者服务器的时候，就会每个服务器执行五条
Rocketmq存储队列
在消息中间件每个topic是有4个写和读队列,主要是解决并发性能的问题的
如果只有一个队列，保证线程安全，必须得给队列进行写操作的时候上锁。
多几个队列，降低并发度，等待时间就短一些。
为什么是四个队列?
因为大多数服务器只有四核，意味着同时最多只能有CPU同时工作
广播模式
在消费模式为集群的情况下，如果机器是集群的，消费是会给集群中的所有机器所消费到
public class Consumer {
public static void main(String[] args) throws Exception {
//定义消息消费者(在同一个JVM中,消费者的组名不能重复)
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("helloConsumerGroup");
//设置nameServer地址
consumer.setNamesrvAddr("43.143.161.59:9876");
//设置订阅的主题
consumer.subscribe("helloTopic","*");
//设置消费模式
consumer.setMessageModel(MessageModel.BROADCASTING);
//设置消息的监听器
consumer.setMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
for(MessageExt msg:list){
String s = new String(msg.getBody(), Charset.defaultCharset());
System.out.println("线程:"+Thread.currentThread()+",消息的内容:"+s);
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
//启动消费者
consumer.start();
}
}
运行结果：
生产者发送了十条消息之后，消费者集群的每个服务器均收到十条数据
顺序消费
实现生产顺序:12345消费顺序12345哪些消息要实现顺序消费,就要让那些消息进入到同一个队列当中,对于消费者来说,一个队列对于一个线程
假设我们没有实现顺序消费的时候
创建生产者
1.创建实体类
@Setter
@Getter
public class OrderStep {
private long orderId;
private String desc;
@Override
public String toString() {
return "OrderStep{" +
"orderId=" + orderId +
", desc='" + desc + '\'' +
'}';
}
}
2.创建测试类
public class OrderUtil {
public static List<OrderStep> buildOrders(){
List<OrderStep> orderList = new ArrayList<OrderStep>();
OrderStep orderDemo = new OrderStep();
orderDemo.setOrderId(15103111039L);
orderDemo.setDesc("创建");
orderList.add(orderDemo);
orderDemo = new OrderStep();
orderDemo.setOrderId(15103111065L);
orderDemo.setDesc("创建");
orderList.add(orderDemo);
orderDemo = new OrderStep();
orderDemo.setOrderId(15103111039L);
orderDemo.setDesc("付款");
orderList.add(orderDemo);
orderDemo = new OrderStep();
orderDemo.setOrderId(15103117235L);
orderDemo.setDesc("创建");
orderList.add(orderDemo);
orderDemo = new OrderStep();
orderDemo.setOrderId(15103111065L);
orderDemo.setDesc("付款");
orderList.add(orderDemo);
orderDemo = new OrderStep();
orderDemo.setOrderId(15103117235L);
orderDemo.setDesc("付款");
orderList.add(orderDemo);
orderDemo = new OrderStep();
orderDemo.setOrderId(15103111065L);
orderDemo.setDesc("完成");
orderList.add(orderDemo);
orderDemo = new OrderStep();
orderDemo.setOrderId(15103111039L);
orderDemo.setDesc("推送");
orderList.add(orderDemo);
orderDemo = new OrderStep();
orderDemo.setOrderId(15103117235L);
orderDemo.setDesc("完成");
orderList.add(orderDemo);
orderDemo = new OrderStep();
orderDemo.setOrderId(15103111039L);
orderDemo.setDesc("完成");
orderList.add(orderDemo);
return orderList;
}
}
3.创建生产者类
public class Producer {
public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("orderlyProducerGroup");
producer.setNamesrvAddr("43.143.161.59:9876");
producer.start();
String topic = "orderTopic";
List<OrderStep> orderSteps = OrderUtil.buildOrders();
for(OrderStep step:orderSteps){
Message msg = new Message(topic,step.toString().getBytes(Charset.defaultCharset()));
producer.sendOneway(msg);
}
producer.shutdown();
}
}
创建消费者类
public class Consumer {
public static void main(String[] args) throws Exception {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("orderlyConsumerGroup");
consumer.setNamesrvAddr("43.143.161.59:9876");
consumer.subscribe("orderTopic","*");
consumer.setMessageModel(MessageModel.BROADCASTING);
consumer.setMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
for(MessageExt msg:list){
String s = new String(msg.getBody(), Charset.defaultCharset());
System.out.println("线程:"+Thread.currentThread()+",消息的内容:"+s);
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
//启动消费者
consumer.start();
}
}
运行结果：
可以看出和我们生产数据的顺序完全不同，整个订单的顺序都反了
如何改实现顺序消费
生产者类
public class Producer {
public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("orderlyProducerGroup");
producer.setNamesrvAddr("43.143.161.59:9876");
producer.start();
String topic = "orderTopic";
List<OrderStep> orderSteps = OrderUtil.buildOrders();
//设置队列选择器
MessageQueueSelector selector = new MessageQueueSelector() {
@Override
public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
System.out.println("队列个数"+list.size());
Long orderId = (Long) o;
int index = (int)(orderId % list.size());
return list.get(index);
}
};
for(OrderStep step:orderSteps){
Message msg = new Message(topic,step.toString().getBytes(Charset.defaultCharset()));
//指定消息选择器,换入的参数
producer.send(msg,selector,step.getOrderId());
}
producer.shutdown();
}
}
消费者类
public class Consumer {
public static void main(String[] args) throws Exception {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("orderlyConsumerGroup");
consumer.setNamesrvAddr("43.143.161.59:9876");
consumer.subscribe("orderTopic","*");
//从什么地方开始消费
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
//一个队列对应一个线程
consumer.setMessageListener(new MessageListenerOrderly() {
@Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
for(MessageExt msg:list){
System.out.println("当前线程："+Thread.currentThread()+":,队列ID"+msg.getQueueId()+",消息内容："+new String(msg.getBody(),Charset.defaultCharset()));
}
return ConsumeOrderlyStatus.SUCCESS;
}
});
//启动消费者
consumer.start();
}
}