淘宝天猫做网站咨询简单代码编程教学

淘宝天猫做网站咨询,简单代码编程教学,91号卡分销平台,如何仿制手机版网站Rocketmq和KafKa类似#xff08;实质上#xff0c;最早的Rocketmq 就是KafKa 的Java版本#xff09;#xff0c;一条消息从生产到被消费#xff0c;将会经历三个阶段#xff1a; 生产阶段#xff0c;Producer 新建消息#xff0c;而后经过网络将消息投递给 MQ Broker。…Rocketmq和KafKa类似实质上最早的Rocketmq 就是KafKa 的Java版本一条消息从生产到被消费将会经历三个阶段生产阶段Producer 新建消息而后经过网络将消息投递给 MQ Broker。这个发送可能会发生丢失比如网络延迟不可达等。存储阶段消息将会存储在 Broker 端磁盘中Broker 根据刷盘策略持久化到硬盘中刚收到Producer的消息在内存中了但是如果Broker 异常宕机了导致消息丢失。消费阶段 Consumer 将会从 Broker 拉取消息以上任一阶段, 都可能会丢失消息只要这三个阶段0丢失就能够完全解决消息丢失的问题。生产阶段如何实现0丢失方式生产阶段有三种send方法:同步发送异步发送单向发送。三种send方法的 客户端api具体如下/** * {link org.apache.rocketmq.client.producer.DefaultMQProducer} */// 同步发送publicSendResultsend(Messagemsg)throwsMQClientException,RemotingException,MQBrokerException,InterruptedException{}// 异步发送sendCallback作为回调publicvoidsend(Messagemsg,SendCallbacksendCallback)throwsMQClientException,RemotingException,InterruptedException{}// 单向发送不关心发送结果最不靠谱publicvoidsendOneway(Messagemsg)throwsMQClientException,RemotingException,InterruptedException{}produce要想发消息时保证消息不丢失可以采用同步发送的方式去发消息send消息方法只要不抛出异常就代表发送成功。发送成功会有多个SendResult 状态以下对每个状态进行说明SEND_OK消息发送成功Broker刷盘、主从同步成功FLUSH_DISK_TIMEOUT消息发送成功但是服务器同步刷盘默认为异步刷盘超时默认超时时间5秒FlUSH_SLAVE_TIMEOUT消息发送成功但是服务器同步复制默认为异步复制到Slave时超时默认超时时间5秒SLAVE_NOT_AVAILABLEBroker从节点不存在注意同步发送只要返回以上四种状态就代表该消息在生产阶段消息正确的投递到了RocketMq生产阶段没有丢失。如果业务要求严格可以使用同步发送并且只取SEND_OK标识消息发送成功是同步发送还是异步发送AP 和 CP 是天然的矛盾 到底是 CP 还是 AP的 需要权衡同步发送的方式 是 CP 高可靠但是性能低。异步发送的方式 是 AP 低可靠但是性能高。为了高可靠CP可以采取同步发送的方式进行发送消息发消息的时候会同步阻塞等待broker返回的结果如果没成功则不会收到SendResult这种是最可靠的。其次是异步发送再回调方法里可以得知是否发送成功。最后单向发送OneWay是最不靠谱的一种发送方式我们无法保证消息真正可达。当然具体的如何选择高可用方案还是要看业务。为了确保万无一失可以选择异步发送 业务维度的 终极0丢失保护措施 实现消息的0丢失。生产端的失败重试策略发送消息如果失败或者超时了则会自动重试。同步发送默认是重试三次可以根据api进行更改比如改为10次producer.setRetryTimesWhenSendFailed(10);其他模式是重试1次具体请参见源码/** * {link org.apache.rocketmq.client.producer.DefaultMQProducer#sendDefaultImpl(Message, CommunicationMode, SendCallback, long)} */// 自动重试次数this.defaultMQProducer.getRetryTimesWhenSendFailed()默认为2如果是同步发送默认重试3次否则重试1次inttimesTotalcommunicationModeCommunicationMode.SYNC?1this.defaultMQProducer.getRetryTimesWhenSendFailed():1;inttimes0;for(;timestimesTotal;times){// 选择发送的消息queueMessageQueuemqSelectedthis.selectOneMessageQueue(topicPublishInfo,lastBrokerName);if(mqSelected!null){try{// 真正的发送逻辑sendKernelImpl。sendResultthis.sendKernelImpl(msg,mq,communicationMode,sendCallback,topicPublishInfo,timeout-costTime);switch(communicationMode){caseASYNC:returnnull;caseONEWAY:returnnull;caseSYNC:// 如果发送失败了则continue意味着还会再次进入for继续重试发送if(sendResult.getSendStatus()!SendStatus.SEND_OK){if(this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()){continue;}}// 发送成功的话将发送结果返回给调用者returnsendResult;default:break;}}catch(RemotingExceptione){continue;}catch(...){continue;}}}上面的核心逻辑中调用sendKernelImpl真正的去发送消息通过核心的发送逻辑可以看出如下同步发送场景的重试次数是1this.defaultMQProducer.getRetryTimesWhenSendFailed()3其他方式默认1次。this.defaultMQProducer.getRetryTimesWhenSendFailed()默认是2我们可以手动设置producer.setRetryTimesWhenSendFailed(10);如果是同步发送sync且发送失败了则continue意味着还会再次进入for继续重试发送同步模式下可以设置严格的消息重试机制比如设置 RetryTimes为一个较大的值如10。当出现网络的瞬时抖动时消息发送可能会失败retries 较大能够自动重试消息发送避免消息丢失。Broker端保证消息不丢失的方法首先想说正常情况下只要 Broker 在正常运行就不会出现丢失消息的问题。但是如果 Broker 出现了故障比如进程死掉了或者服务器宕机了还是可能会丢失消息的。如果确保万无一失实现Broker端保证消息不丢失有两板斧Broker端第一板斧设置严格的副本同步机制Broker端第二板斧设置严格的消息刷盘机制Broker端第一板斧设置严格的副本同步机制RocketMQ 通过多副本机制来解决的高可用核心思想也挺简单的如果数据保存在一台机器上你觉得可靠性不够那么我就把相同的数据保存到多台机器上某台机器宕机了可以由其它机器提供相同的服务和数据。首先Broker需要集群部署通过主从模式包括 topic 数据的高可用。为了消息0丢失可以配置设置严格的副本同步机制等Master 把消息同步给 Slave后才去通知Producer说消息ok。设置严格的副本同步机制 RocketMQ 修改broker刷盘配置如下所以我们还可以配置不仅是等Master刷完盘就通知Producer而是等Master和Slave都刷完盘后才去通知Producer说消息ok了。## 默认为ASYNC_MASTERbrokerRoleSYNC_MASTERBroker端第二板斧设置严格的消息刷盘机制RocketMQ持久化消息分为两种同步刷盘和异步刷盘。RocketMQ和kafka一样的刷盘的方式有同步刷盘和异步刷盘两种。同步刷盘指的是生产者消息发过来时只有持久化到磁盘RocketMQ、kafka的存储端Broker才返回一个成功的ACK响应这就是同步刷盘。它保证消息不丢失但是影响了性能。异步刷盘指的是消息写入PageCache缓存就返回一个成功的ACK响应不管消息有没有落盘就返回一个成功的ACK响应。这样提高了MQ的性能但是如果这时候机器断电了就会丢失消息。同步刷盘和异步刷盘的区别如下:同步刷盘:当数据写如到内存中之后立刻刷盘(同步)在保证刷盘成功的前提下响应client。异步刷盘:数据写入内存后直接响应client。异步将内存中的数据持久化到磁盘上。同步刷盘和异步输盘的优劣:同步刷盘保证了数据的可靠性,保证数据不会丢失。同步刷盘效率较低,因为client获取响应需要等待刷盘时间为了提升效率通常采用批量输盘的方式每次刷盘将会flush内存中的所有数据。(若底层的存储为mmap则每次刷盘将刷新所有的dirty页)异步刷盘不能保证数据的可靠性.异步刷盘可以提高系统的吞吐量.常见的异步刷盘方式有两种,分别是定时刷盘和触发式刷盘。定时刷盘可设置为如每1s刷新一次内存.触发刷盘为当内存中数据到达一定的值会触发异步刷盘程序进行刷盘。Broker端第二板斧设置严格的消息刷盘机制设置为Kafka同步刷盘。RocketMQ默认情况是异步刷盘Broker收到消息后会先存到cache里然后通知Producer说消息我收到且存储成功。Broker起个线程异步的去持久化到磁盘中但是Broker还没持久化到磁盘就宕机的话消息就丢失了。同步刷盘的话是收到消息存到cache后并不会通知Producer说消息已经ok了而是会等到持久化到磁盘中后才会通知Producer说消息完事了。同步刷盘的方式 是 CP 高可靠但是性能低。异步刷盘的方式 是 AP 低可靠但是性能高。为了高可靠CP可以采取同步刷盘保障了消息不会丢失但是性能不如异步高。如何设置RocketMQ同步刷盘RocketMQ 修改broker刷盘配置如下## 默认情况为 ASYNC_FLUSH修改为同步刷盘SYNC_FLUSH实际场景看业务同步刷盘效率肯定不如异步刷盘高。 flushDiskTypeSYNC_FLUSH对应的RocketMQ源码类如下packageorg.apache.rocketmq.store.config;publicenumFlushDiskType{// 同步刷盘SYNC_FLUSH,// 异步刷盘默认ASYNC_FLUSH}异步刷盘默认10s执行一次源码如下/* * {link org.apache.rocketmq.store.CommitLog#run()} */while(!this.isStopped()){try{// 等待10sthis.waitForRunning(10);// 刷盘this.doCommit();}catch(Exceptione){CommitLog.log.warn(this.getServiceName() service has exception. ,e);}}Broker端0丢失的配置总结Broker端的配置若想很严格的保证Broker存储消息阶段消息不丢失则需要如下配置# master 节点配置 flushDiskTypeSYNC_FLUSHbrokerRoleSYNC_MASTER # slave 节点配置 brokerRoleslave flushDiskTypeSYNC_FLUSH上面这个配置含义是Producer发消息到Broker后Broker的Master节点先持久化到磁盘中然后同步数据给Slave节点Slave节点同步完且落盘完成后才会返回给Producer说消息ok了。严格的消息刷盘机制 严格的消息同步机制能够确保 Broker端保证消息不丢失Consumer消费者保证消息不丢失的方法如果要保证 Consumer消费者0 丢失 Consumer 端的策略是啥呢普通的情况下rocketMq拉取消息后执行业务逻辑。一旦Consumer执行成功将会返回一个ACK响应给 Broker这时MQ就会修改offset将该消息标记为已消费不再往其他消费者推送消息。如果出现消费超时(默认15分钟)、拉取消息后消费者服务宕机等消费失败的情况此时的Broker由于没有等到消费者返回的ACK会向同一个消费者组中的其他消费者间隔性的重发消息直到消息返回成功默认是重复发送16次若16次还是没有消费成功那么该消息会转移到死信队列,人工处理或是单独写服务处理这些死信消息但是 消费者也有两种消费模式同步消费消费线程完成业务操作异步消息 独立业务线程池 完成业务操作一般大家在rocketMq 在并发消费模式下这个模式默认有20个消费线程如何保证客户端的高可用两种场景同步消费场景业务代码手动发送CONSUME_SUCCESS 保证 消息者的0丢失异步消费场景需要通过业务维度的 终极0丢失保护措施本地消息表定时扫描 保证 消息者的0丢失1、同步消费发送CONSUME_SUCCESS同步消费指的是拉取消息的线程先把消息拉取到本地然后进行业务逻辑业务逻辑完成后手动进行ack确认这时候才会真正的代表消费完成。举个例子consumer.registerMessageListener(newMessageListenerConcurrently(){OverridepublicConsumeConcurrentlyStatusconsumeMessage(ListMessageExtmsgs,ConsumeConcurrentlyContextconsumeConcurrentlyContext){for(MessageExtmsg:msgs){StringstrnewString(msg.getBody());// 消费者 线程 同步进行 业务处理System.out.println(str);}// ack只有等上面一系列逻辑都处理完后// 发 CONSUME_SUCCESS才会通知broker说消息消费完成// 如果上面发生异常没有走到这步ack则消息还是未消费状态。returnConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});2、异步消费场景如何保证0丢失rocketMq 在并发消费模式下默认有20个消费线程但是这个还是有限制的。如果实现高性能呢可以一方面去进行线程扩容 比如通过修改配置扩容到100个rocketMq 同步消费线程但是这个会在没有 活儿干的场景浪费宝贵的CPU资源。可以另一方便通过异步的 可动态扩容的业务专用线程池去完成 消费的业务处理。那么如果是设置了业务的专用线程池则需要通过业务维度的 终极0丢失保护措施本地消息表定时扫描 保证 消息者的0丢失业务维度的 终极0丢失保护措施本地消息表定时扫描慎重提示前面三板斧并不能保证100%的0丢失。因为百密一疏任何环节的异常都有可能导致数据丢失。有没有业务维度的 终极保护措施呢有本地消息表定时扫描本地消息表定时扫描 方案和本地消息表事务机制类似也是采用 本地消息表定时扫描 相结合的架构方案。大概流程如下图1、设计一个本地消息表可以存储在DB里或者其它存储引擎里用户保存消息的消费状态2、Producer 发送消息之前首先保证消息的发生状态并且初始化为待发送3、如果消费者如库存服务完成的消费则通过RPC调用Producer 去更新一下消息状态4、Producer 利用定时任务扫描 过期的消息比如10分钟到期再次进行发送。在这里想说的是本地消息表定时扫描 的架构方案 是业务层通过额外的机制来保证消息数据发送的完整性是一种很重的方案。这个方案的两个特点CP 不是 AP性能低需要 做好幂等性设计如何降低业务维度的 终极0丢失保护措施带来的性能耗损可以减少本地消息表的规模对于正常投递的消息不做跟踪只把生产端发送失败的消息、消费端消费失败的消息记录到数据库并启动一个定时任务扫描发送失败的消息重新发送直到超过阈值如10次超过之后发送邮件或短信通知人工介入处理。RocketMQ的0丢失的最佳实践1.Producer端使用同步发送方式发送消息可以提高可靠性。记住如果使用带有回调通知的 send异步 方法发送去提高性能则需要结合 本地消息表定时扫描 的架构去实现业务维度的 高可靠。2.Producer端同步模式下可以设置严格的消息重试机制比如设置 RetryTimes为一个较大的值如10。当出现网络的瞬时抖动时消息发送可能会失败retries 较大能够自动重试消息发送避免消息丢失。。3.Broker 端设置严格的副本同步机制 。4.Broker 端设置严格的消息刷盘机制。5.Consumer 端确保消息消费完成再提交。可以使用同步消费并发送CONSUME_SUCCESS。6.业务维度的的0丢失架构采用 本地消息表定时扫描 架构方案实现业务维度的 0丢失100%可靠性。