现在网站开发模式,南阳做网站 汉狮公司,如何建设医院网站,成都网站开发公司有哪些引言
Apache RocketMQ 诞生至今#xff0c;历经十余年大规模业务稳定性打磨#xff0c;服务了阿里集团内部业务以及阿里云数以万计的企业客户。作为金融级可靠的业务消息方案#xff0c;RocketMQ 从创建之初就一直专注于业务集成领域的异步通信能力构建。本篇将继续业务消息…引言
Apache RocketMQ 诞生至今历经十余年大规模业务稳定性打磨服务了阿里集团内部业务以及阿里云数以万计的企业客户。作为金融级可靠的业务消息方案RocketMQ 从创建之初就一直专注于业务集成领域的异步通信能力构建。本篇将继续业务消息集成的场景从功能原理、应用案例、最佳实践以及实战等角度介绍 RocketMQ 的顺序消息功能。
简介
顺序消息是消息队列 RocketMQ 版提供的一种对消息发送和消费顺序有严格要求的消息。对于一个指定的 Topic同一 MessageGroup 的消息按照严格的先进先出FIFO原则进行发布和消费即先发布的消息先消费后发布的消息后消费服务端严格按照发送顺序进行存储、消费。同一 MessageGroup 的消息保证顺序不同 MessageGroup 之间的消息顺序不做要求因此需做到两点发送的顺序性和消费的顺序性。 功能原理
在这里首先抛出一个问题在日常的接触中许多 RocketMQ 使用者会认为既然顺序消息能在普通消息的基础上实现顺序看起来就是普通消息的加强版那么为什么不全部都使用顺序消息呢接下来就会围绕这个问题对比普通消息和顺序消息进行阐述。
顺序发送
在分布式环境下保证消息的全局顺序性是十分困难的例如两个 RocketMQ Producer A 与 Producer B它们在没有沟通的情况下各自向 RocketMQ 服务端发送消息 a 和消息 b由于分布式系统的限制我们无法保证 a 和 b 的顺序。因此业界消息系统通常保证的是分区的顺序性即保证带有同一属性的消息的顺序我们将该属性称之为 MessageGroup。如图所示ProducerA 发送了 MessageGroup 属性为 A 的两条消息 A1A2 和 MessageGroup 属性为 B 的 B1B2而 ProducerB 发送了 MessageGroup 属性为 C 的两条属性 C1C2。 同时对于同一 MessageGroup为了保证其发送顺序的先后性比较简单的做法是构造一个单线程的场景即不同的 MessageGroup 由不同的 Producer 负责并且对于每一个 Producer 而言顺序消息是同步发送的。同步发送的好处是显而易见的在客户端得到上一条消息的发送结果后再发送下一条即能准确保证发送顺序若使用异步发送或多线程则很难保证这一点。 因此可以看到虽然在底层原理上顺序消息发送和普通消息发送并无二异但是为了保证顺序消息的发送顺序性同步发送的方式相比较普通消息实际上降低了消息的最大吞吐。
顺序消费
与顺序消息不同的是普通消息的消费实际上没有任何限制消费者拉取的消息是被异步、并发消费的而顺序消息需要保证对于同一个 MessageGroup同一时刻只有一个客户端在消费消息并且在该条消息被确认消费完成之前或者进入死信队列消费者无法消费同一 MessageGroup 的下一条消息否则消费的顺序性将得不到保证。因此这里存在着一个消费瓶颈该瓶颈取决于用户自身的业务处理逻辑。极端情况下当某一 MessageGroup 的消息过多时就可能导致消费堆积。当然也需要明确的是这里的语境都指的是同一 MessageGroup不同 MessageGroup 的消息之间并不存在顺序性的关联是可以进行并发消费的。因此全文中提到的顺序实际上是一种偏序。 小结
无论对于发送还是消费我们通过 MessageGroup 的方式将消息分组即并发的基本单元是 MessageGroup不同的 MessageGroup 可以并发的发送和消费从而一定程度具备了可拓展性支持多队列存储、水平拆分、并发消费且不受影响。回顾普通消息站在顺序消息的视角可以认为普通消息的并发基本单元是单条消息即每条消息均拥有不同的 MessageGroup。
我们回到开头那个问题
既然顺序消息能在普通消息的基础上实现顺序看起来就是普通消息的加强版那么为什么不全部都使用顺序消息呢现在大家对于这个问题可能有一个基本的印象了消息的顺序性当然很好但是为了实现顺序性也是有代价的。
下述是一个表格简要对比了顺序消息和普通消息。 最佳实践
合理设置 MessageGroup
MessageGroup 会有很多错误的选择以某电商平台为例某电商平台将商家 ID 作为 MessageGroup因为部分规模较大的商家会产出较多订单由于下游消费能力的限制因此这部分商家所对应的订单就发生了严重的堆积。正确的做法应当是将订单号作为 MessageGroup而且站在背后的业务逻辑上来说同一订单才有顺序性的要求。即选择 MessageGroup 的最佳实践是MessageGroup 生命周期最好较为短暂且不同 MessageGroup 的数量应当尽量相同且均匀。
同步发送和发送重试
如之前章节所述需使用同步发送和发送重试来保证发送的顺序性。
消费幂等
消息传输链路在异常场景下会有少量重复业务消费是需要做消费幂等避免重复处理带来的风险。
应用案例
用户注册需要发送验证码以用户 ID 作为 MessageGroup那么同一个用户发送的消息都会按照发布的先后顺序来消费。电商的订单创建以订单 ID 作为 MessageGroup那么同一个订单相关的创建订单消息、订单支付消息、订单退款消息、订单物流消息都会按照发布的先后顺序来消费。实战
发送
可以看到该发送案例设置了 MessageGroup 并且使用了同步发送发送的代码如下
public class ProducerFifoMessageExample {private static final Logger LOGGER LoggerFactory.getLogger(ProducerFifoMessageExample.class);private ProducerFifoMessageExample() {}public static void main(String[] args) throws ClientException, IOException {final ClientServiceProvider provider ClientServiceProvider.loadService();// Credential provider is optional for client configuration.String accessKey yourAccessKey;String secretKey yourSecretKey;SessionCredentialsProvider sessionCredentialsProvider new StaticSessionCredentialsProvider(accessKey, secretKey);String endpoints foobar.com:8080;ClientConfiguration clientConfiguration ClientConfiguration.newBuilder().setEndpoints(endpoints).setCredentialProvider(sessionCredentialsProvider).build();String topic yourFifoTopic;final Producer producer provider.newProducerBuilder().setClientConfiguration(clientConfiguration)// Set the topic name(s), which is optional. It makes producer could prefetch the topic route before // message publishing..setTopics(topic)// May throw {link ClientException} if the producer is not initialized..build();// Define your message body.byte[] body This is a FIFO message for Apache RocketMQ.getBytes(StandardCharsets.UTF_8);String tag yourMessageTagA;final Message message provider.newMessageBuilder()// Set topic for the current message..setTopic(topic)// Message secondary classifier of message besides topic..setTag(tag)// Key(s) of the message, another way to mark message besides message id..setKeys(yourMessageKey-1ff69ada8e0e)// Message group decides the message delivery order..setMessageGroup(youMessageGroup0).setBody(body).build();try {final SendReceipt sendReceipt producer.send(message);LOGGER.info(Send message successfully, messageId{}, sendReceipt.getMessageId());} catch (Throwable t) {LOGGER.error(Failed to send message, t);}// Close the producer when you dont need it anymore.producer.close();}
}
消费
消费的代码如下
public class SimpleConsumerExample {private static final Logger LOGGER LoggerFactory.getLogger(SimpleConsumerExample.class);private SimpleConsumerExample() {}public static void main(String[] args) throws ClientException, IOException {final ClientServiceProvider provider ClientServiceProvider.loadService();// Credential provider is optional for client configuration.String accessKey yourAccessKey;String secretKey yourSecretKey;SessionCredentialsProvider sessionCredentialsProvider new StaticSessionCredentialsProvider(accessKey, secretKey);String endpoints foobar.com:8080;ClientConfiguration clientConfiguration ClientConfiguration.newBuilder().setEndpoints(endpoints).setCredentialProvider(sessionCredentialsProvider).build();String consumerGroup yourConsumerGroup;Duration awaitDuration Duration.ofSeconds(30);String tag yourMessageTagA;String topic yourTopic;FilterExpression filterExpression new FilterExpression(tag, FilterExpressionType.TAG);SimpleConsumer consumer provider.newSimpleConsumerBuilder().setClientConfiguration(clientConfiguration)// Set the consumer group name..setConsumerGroup(consumerGroup)// set await duration for long-polling..setAwaitDuration(awaitDuration)// Set the subscription for the consumer..setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression)).build();// Max message num for each long polling.int maxMessageNum 16;// Set message invisible duration after it is received.Duration invisibleDuration Duration.ofSeconds(5);final ListMessageView messages consumer.receive(maxMessageNum, invisibleDuration);for (MessageView message : messages) {try {consumer.ack(message);} catch (Throwable t) {LOGGER.error(Failed to acknowledge message, messageId{}, message.getMessageId(), t);}}// Close the simple consumer when you dont need it anymore.consumer.close();}
}
作者绍舒
原文链接
本文为阿里云原创内容未经允许不得转载
