安泽网站建设,怎么让别人看到自己做的网站,wordpress 在线点餐,wordpress站点迁移顺序消息的应用场景
在有序事件处理、撮合交易、数据实时增量同步等场景下#xff0c;异构系统间需要维持强一致的状态同步#xff0c;上游的事件变更需要按照顺序传递到下游进行处理。
例如需要保证一个订单的生成、付款和发货#xff0c;这三件事情是被顺序执行的。
如…顺序消息的应用场景
在有序事件处理、撮合交易、数据实时增量同步等场景下异构系统间需要维持强一致的状态同步上游的事件变更需要按照顺序传递到下游进行处理。
例如需要保证一个订单的生成、付款和发货这三件事情是被顺序执行的。
如何消息的顺序性
RocketMQ消息的顺序性分为两部分生产顺序性和消费顺序性只有同时满足了生产顺序性和消费顺序性才能达到消息整体的有序性
生产的顺序性
要保证发送消息的顺序性就必须保证消息以下条件
单一生产者顺序消息必须由单一生产者产生不同生产者分布在不同的系统即使设置相同的分区键不同生产者之间产生的消息也无法判定其先后顺序。串行发送生产者客户端支持多线程安全访问但如果生产者使用多线程并行发送则不同线程间产生的消息将无法判定其先后顺序。
总结单一生产者需要制定消息的顺序性并且需要将顺序消息根据分区键发送到一个队列上在发送时需要使用串行发送
Test
public void sendOrder() throws Exception {DefaultMQProducer producer new DefaultMQProducer(test-producer-group);producer.setNamesrvAddr(RocketMQConfig.NAME_SERVER_ADDR);producer.start();// 分区keyint orderId 1;for (int i 0; i 1000; i) {Message message new Message(RocketMQConfig.TEST_TOPIC, (顺序 i).getBytes(Charset.defaultCharset()));// 发送顺序消息需要传递分区键SendResult sendResult1 producer.send(message, new MessageQueueSelector() {Overridepublic MessageQueue select(ListMessageQueue mqs, Message msg, Object arg) {// 分区键int orderId (int) arg;// 根据分区键取模int index orderId % mqs.size();// 返回要发送到哪个队列中去return mqs.get(index);}}, orderId);System.out.println(sendResult1.getSendStatus());}producer.shutdown();
}消息的顺序性
消费者在push模式下有两种消息方式 MessageListenerOrderly会启动多个线程处理消息但是会加锁实际上会转变为串行进行实现消息的顺序性 顺序消费的结果 MessageListenerConcurrently会启动多个线程处理消费者并且不保证加锁不保证消息的顺序性 非顺序消费的结果 因此为了保证消息的顺序性需要使用MessageListenerOrderly来处理消息
Test
public void consumerOrder() throws Exception{// 1.创建消费端指明消费者属于哪个组DefaultMQPushConsumer consumer new DefaultMQPushConsumer(test_group_order);// 2.注册NameServer地址consumer.setNamesrvAddr(RocketMQConfig.NAME_SERVER_ADDR);// 3.订阅topic并且可以根据标签进行定向消费consumer.subscribe(RocketMQConfig.TEST_TOPIC, *);// 4.注册监听器broker推送消息后处理顺序消息consumer.registerMessageListener(new MessageListenerOrderly() {Overridepublic ConsumeOrderlyStatus consumeMessage(ListMessageExt msgs, ConsumeOrderlyContext context) {System.out.println(Thread.currentThread().getName() 收到的msg大小 msgs.size());for (MessageExt msg : msgs) {String content new String(msg.getBody());System.out.println(Thread.currentThread().getName() 收到的消息体 content);}return ConsumeOrderlyStatus.SUCCESS;}});// 5.启动消费端consumer.start();// 防止主线程退出Thread.sleep(Integer.MAX_VALUE);
}注在验证顺序消息时当这两种消费模式在启动情况下是都可以顺序消费的。 只有先发送消息在启动消费并发消费才会出现乱序的情况 猜测 单生产者进行发送消息并发量不高发送的数据量太低
