珠海企业网站建设报价,鄂州网吧什么时候恢复营业,什么浏览器可以看任何网站,城市建设服务中心网站前些天发现了一个巨牛的人工智能学习网站#xff0c;通俗易懂#xff0c;风趣幽默#xff0c;忍不住分享一下给大家。点击跳转到教程。
分布式消息系统作为实现分布式系统可扩展、可伸缩性的关键组件#xff0c;需要具有高吞吐量、高可用等特点。而谈到消息系统的设计通俗易懂风趣幽默忍不住分享一下给大家。点击跳转到教程。
分布式消息系统作为实现分布式系统可扩展、可伸缩性的关键组件需要具有高吞吐量、高可用等特点。而谈到消息系统的设计就回避不了两个问题 消息的顺序问题消息的重复问题RocketMQ作为阿里开源的一款高性能、高吞吐量的消息中间件它是怎样来解决这两个问题的RocketMQ 有哪些关键特性其实现原理是怎样的
关键特性以及其实现原理
一、顺序消息
消息有序指的是可以按照消息的发送顺序来消费。例如一笔订单产生了 3 条消息分别是订单创建、订单付款、订单完成。消费时要按照顺序依次消费才有意义。与此同时多笔订单之间又是可以并行消费的。首先来看如下示例
假如生产者产生了2条消息M1、M2要保证这两条消息的顺序应该怎样做你脑中想到的可能是这样 你可能会采用这种方式保证消息顺序 假定M1发送到S1M2发送到S2如果要保证M1先于M2被消费那么需要M1到达消费端被消费后通知S2然后S2再将M2发送到消费端。
这个模型存在的问题是如果M1和M2分别发送到两台Server上就不能保证M1先达到MQ集群也不能保证M1被先消费。换个角度看如果M2先于M1达到MQ集群甚至M2被消费后M1才达到消费端这时消息也就乱序了说明以上模型是不能保证消息的顺序的。如何才能在MQ集群保证消息的顺序一种简单的方式就是将M1、M2发送到同一个Server上 保证消息顺序你改进后的方法 这样可以保证M1先于M2到达MQServer生产者等待M1发送成功后再发送M2根据先达到先被消费的原则M1会先于M2被消费这样就保证了消息的顺序。
这个模型也仅仅是理论上可以保证消息的顺序在实际场景中可能会遇到下面的问题 网络延迟问题
只要将消息从一台服务器发往另一台服务器就会存在网络延迟问题。如上图所示如果发送M1耗时大于发送M2的耗时那么M2就仍将被先消费仍然不能保证消息的顺序。即使M1和M2同时到达消费端由于不清楚消费端1和消费端2的负载情况仍然有可能出现M2先于M1被消费的情况。
那如何解决这个问题将M1和M2发往同一个消费者且发送M1后需要消费端响应成功后才能发送M2。
聪明的你可能已经想到另外的问题如果M1被发送到消费端后消费端1没有响应那是继续发送M2呢还是重新发送M1一般为了保证消息一定被消费肯定会选择重发M1到另外一个消费端2就如下图所示。 保证消息顺序的正确姿势
这样的模型就严格保证消息的顺序细心的你仍然会发现问题消费端1没有响应Server时有两种情况一种是M1确实没有到达(数据在网络传送中丢失)另外一种消费端已经消费M1且已经发送响应消息只是MQ Server端没有收到。如果是第二种情况重发M1就会造成M1被重复消费。也就引入了我们要说的第二个问题消息重复问题这个后文会详细讲解。
回过头来看消息顺序问题严格的顺序消息非常容易理解也可以通过文中所描述的方式来简单处理。总结起来要实现严格的顺序消息简单且可行的办法就是 保证生产者 - MQServer - 消费者是一对一对一的关系 这样的设计虽然简单易行但也会存在一些很严重的问题比如 并行度就会成为消息系统的瓶颈吞吐量不够更多的异常处理比如只要消费端出现问题就会导致整个处理流程阻塞我们不得不花费更多的精力来解决阻塞的问题。但我们的最终目标是要集群的高容错性和高吞吐量。这似乎是一对不可调和的矛盾那么阿里是如何解决的 世界上解决一个计算机问题最简单的方法“恰好”不需要解决它—— 沈询 有些问题看起来很重要但实际上我们可以通过合理的设计或者将问题分解来规避。如果硬要把时间花在解决问题本身实际上不仅效率低下而且也是一种浪费。从这个角度来看消息的顺序问题我们可以得出两个结论 不关注乱序的应用实际大量存在队列无序并不意味着消息无序所以从业务层面来保证消息的顺序而不仅仅是依赖于消息系统是不是我们应该寻求的一种更合理的方式
最后我们从源码角度分析RocketMQ怎么实现发送顺序消息。
RocketMQ通过轮询所有队列的方式来确定消息被发送到哪一个队列负载均衡策略。比如下面的示例中订单号相同的消息会被先后发送到同一个队列中
// RocketMQ通过MessageQueueSelector中实现的算法来确定消息发送到哪一个队列上
// RocketMQ默认提供了两种MessageQueueSelector实现随机/Hash
// 当然你可以根据业务实现自己的MessageQueueSelector来决定消息按照何种策略发送到消息队列中
SendResult sendResult producer.send(msg, new MessageQueueSelector() {Overridepublic MessageQueue select(ListMessageQueue mqs, Message msg, Object arg) {Integer id (Integer) arg;int index id % mqs.size();return mqs.get(index);}
}, orderId);在获取到路由信息以后会根据MessageQueueSelector实现的算法来选择一个队列同一个OrderId获取到的肯定是同一个队列。
private SendResult send() {// 获取topic路由信息TopicPublishInfo topicPublishInfo this.tryToFindTopicPublishInfo(msg.getTopic());if (topicPublishInfo ! null topicPublishInfo.ok()) {MessageQueue mq null;// 根据我们的算法选择一个发送队列// 这里的arg orderIdmq selector.select(topicPublishInfo.getMessageQueueList(), msg, arg);if (mq ! null) {return this.sendKernelImpl(msg, mq, communicationMode, sendCallback, timeout);}}
}二、消息重复
上面在解决消息顺序问题时引入了一个新的问题就是消息重复。那么RocketMQ是怎样解决消息重复的问题呢还是“恰好”不解决。
造成消息重复的根本原因是网络不可达。只要通过网络交换数据就无法避免这个问题。所以解决这个问题的办法就是绕过这个问题。那么问题就变成了如果消费端收到两条一样的消息应该怎样处理 消费端处理消息的业务逻辑保持幂等性保证每条消息都有唯一编号且保证消息处理成功与去重表的日志同时出现第1条很好理解只要保持幂等性不管来多少条重复消息最后处理的结果都一样。第2条原理就是利用一张日志表来记录已经处理成功的消息的ID如果新到的消息ID已经在日志表中那么就不再处理这条消息。
第1条解决方案很明显应该在消费端实现不属于消息系统要实现的功能。第2条可以消息系统实现也可以业务端实现。正常情况下出现重复消息的概率其实很小如果由消息系统来实现的话肯定会对消息系统的吞吐量和高可用有影响所以最好还是由业务端自己处理消息重复的问题这也是RocketMQ不解决消息重复的问题的原因。
RocketMQ不保证消息不重复如果你的业务需要保证严格的不重复消息需要你自己在业务端去重。
三、事务消息
RocketMQ除了支持普通消息顺序消息另外还支持事务消息。首先讨论一下什么是事务消息以及支持事务消息的必要性。我们以一个转帐的场景为例来说明这个问题Bob向Smith转账100块。
在单机环境下执行事务的情况大概是下面这个样子 单机环境下转账事务示意图
当用户增长到一定程度Bob和Smith的账户及余额信息已经不在同一台服务器上了那么上面的流程就变成了这样 集群环境下转账事务示意图
这时候你会发现同样是一个转账的业务在集群环境下耗时居然成倍的增长这显然是不能够接受的。那如何来规避这个问题 大事务 小事务 异步 将大事务拆分成多个小事务异步执行。这样基本上能够将跨机事务的执行效率优化到与单机一致。转账的事务就可以分解成如下两个小事务 小事务异步消息 图中执行本地事务Bob账户扣款和发送异步消息应该保证同时成功或者同时失败也就是扣款成功了发送消息一定要成功如果扣款失败了就不能再发送消息。那问题是我们是先扣款还是先发送消息呢
首先看下先发送消息的情况大致的示意图如下 事务消息先发送消息
存在的问题是如果消息发送成功但是扣款失败消费端就会消费此消息进而向Smith账户加钱。
先发消息不行那就先扣款吧大致的示意图如下 事务消息-先扣款
存在的问题跟上面类似如果扣款成功发送消息失败就会出现Bob扣钱了但是Smith账户未加钱。
可能大家会有很多的方法来解决这个问题比如直接将发消息放到Bob扣款的事务中去如果发送失败抛出异常事务回滚。这样的处理方式也符合“恰好”不需要解决的原则。 这里需要说明一下如果使用Spring来管理事物的话大可以将发送消息的逻辑放到本地事物中去发送消息失败抛出异常Spring捕捉到异常后就会回滚此事物以此来保证本地事物与发送消息的原子性。 RocketMQ支持事务消息下面来看看RocketMQ是怎样来实现的。 RocketMQ实现发送事务消息
RocketMQ第一阶段发送Prepared消息时会拿到消息的地址第二阶段执行本地事物第三阶段通过第一阶段拿到的地址去访问消息并修改消息的状态。
细心的你可能又发现问题了如果确认消息发送失败了怎么办RocketMQ会定期扫描消息集群中的事物消息如果发现了Prepared消息它会向消息发送端(生产者)确认Bob的钱到底是减了还是没减呢如果减了是回滚还是继续发送确认消息呢RocketMQ会根据发送端设置的策略来决定是回滚还是继续发送确认消息。这样就保证了消息发送与本地事务同时成功或同时失败。
那我们来看下RocketMQ源码是如何处理事务消息的。客户端发送事务消息的部分完整代码请查看rocketmq-example工程下的com.alibaba.rocketmq.example.transaction.TransactionProducer
// 发送事务消息的一系列准备工作
// 未决事务MQ服务器回查客户端
// 也就是上文所说的当RocketMQ发现Prepared消息时会根据这个Listener实现的策略来决断事务
TransactionCheckListener transactionCheckListener new TransactionCheckListenerImpl();
// 构造事务消息的生产者
TransactionMQProducer producer new TransactionMQProducer(groupName);
// 设置事务决断处理类
producer.setTransactionCheckListener(transactionCheckListener);
// 本地事务的处理逻辑相当于示例中检查Bob账户并扣钱的逻辑
TransactionExecuterImpl tranExecuter new TransactionExecuterImpl();
producer.start()
// 构造MSG省略构造参数
Message msg new Message(......);
// 发送消息
SendResult sendResult producer.sendMessageInTransaction(msg, tranExecuter, null);
producer.shutdown();接着查看sendMessageInTransaction方法的源码总共分为3个阶段发送Prepared消息、执行本地事务、发送确认消息。
// 事务消息的发送过程
public TransactionSendResult sendMessageInTransaction(.....) {// 逻辑代码非实际代码// 1.发送消息sendResult this.send(msg);// sendResult.getSendStatus() SEND_OK// 2.如果消息发送成功处理与消息关联的本地事务单元LocalTransactionState localTransactionState tranExecuter.executeLocalTransactionBranch(msg, arg);// 3.结束事务this.endTransaction(sendResult, localTransactionState, localException);
}endTransaction方法会将请求发往broker(mq server)去更新事务消息的最终状态
根据sendResult找到Prepared消息 sendResult包含事务消息的ID根据localTransaction更新消息的最终状态
如果endTransaction方法执行失败数据没有发送到broker导致事务消息的 状态更新失败broker会有回查线程定时默认1分钟扫描每个存储事务状态的表格文件如果是已经提交或者回滚的消息直接跳过如果是prepared状态则会向Producer发起CheckTransaction请求Producer会调用DefaultMQProducerImpl.checkTransactionState()方法来处理broker的定时回调请求而checkTransactionState会调用我们的事务设置的决断方法来决定是回滚事务还是继续执行最后调用endTransactionOneway让broker来更新消息的最终状态。
再回到转账的例子如果Bob的账户的余额已经减少且消息已经发送成功Smith端开始消费这条消息这个时候就会出现消费失败和消费超时两个问题解决超时问题的思路就是一直重试直到消费端消费消息成功整个过程中有可能会出现消息重复的问题按照前面的思路解决即可。 消费事务消息
这样基本上可以解决消费端超时问题但是如果消费失败怎么办阿里提供给我们的解决方法是人工解决。大家可以考虑一下按照事务的流程因为某种原因Smith加款失败那么需要回滚整个流程。如果消息系统要实现这个回滚流程的话系统复杂度将大大提升且很容易出现Bug估计出现Bug的概率会比消费失败的概率大很多。这也是RocketMQ目前暂时没有解决这个问题的原因在设计实现消息系统时我们需要衡量是否值得花这么大的代价来解决这样一个出现概率非常小的问题这也是大家在解决疑难问题时需要多多思考的地方。 20160321补充在3.2.6版本中移除了事务消息的实现所以此版本不支持事务消息具体情况请参考rocketmq的issues(已失效)https://github.com/alibaba/RocketMQ/issues/65https://github.com/alibaba/RocketMQ/issues/138https://github.com/alibaba/RocketMQ/issues/156 四、Producer如何发送消息
Producer轮询某topic下的所有队列的方式来实现发送方的负载均衡如下图所示 producer发送消息负载均衡 首先分析一下RocketMQ的客户端发送消息的源码 // 构造Producer
DefaultMQProducer producer new DefaultMQProducer(ProducerGroupName);
// 初始化Producer整个应用生命周期内只需要初始化1次
producer.start();
// 构造Message
Message msg new Message(TopicTest1,// topicTagA,// tag给消息打标签,用于区分一类消息可为nullOrderID188,// key自定义Key可以用于去重可为null(Hello MetaQ).getBytes());// body消息内容
// 发送消息并返回结果
SendResult sendResult producer.send(msg);
// 清理资源关闭网络连接注销自己
producer.shutdown();在整个应用生命周期内生产者需要调用一次start方法来初始化初始化主要完成的任务有 如果没有指定namesrv地址将会自动寻址启动定时任务更新namesrv地址、从namsrv更新topic路由信息、清理已经挂掉的broker、向所有broker发送心跳...启动负载均衡的服务初始化完成后开始发送消息发送消息的主要代码如下
private SendResult sendDefaultImpl(Message msg,......) {// 检查Producer的状态是否是RUNNINGthis.makeSureStateOK();// 检查msg是否合法是否为null、topic,body是否为空、body是否超长Validators.checkMessage(msg, this.defaultMQProducer);// 获取topic路由信息TopicPublishInfo topicPublishInfo this.tryToFindTopicPublishInfo(msg.getTopic());// 从路由信息中选择一个消息队列MessageQueue mq topicPublishInfo.selectOneMessageQueue(lastBrokerName);// 将消息发送到该队列上去sendResult this.sendKernelImpl(msg, mq, communicationMode, sendCallback, timeout);
}代码中需要关注的两个方法tryToFindTopicPublishInfo和selectOneMessageQueue。前面说过在producer初始化时会启动定时任务获取路由信息并更新到本地缓存所以tryToFindTopicPublishInfo会首先从缓存中获取topic路由信息如果没有获取到则会自己去namesrv获取路由信息。selectOneMessageQueue方法通过轮询的方式返回一个队列以达到负载均衡的目的。
如果Producer发送消息失败会自动重试重试的策略
重试次数 retryTimesWhenSendFailed可配置总的耗时包含重试n次的耗时 sendMsgTimeout发送消息时传入的参数同时满足上面两个条件后Producer会选择另外一个队列发送消息
五、消息存储
RocketMQ的消息存储是由consume queue和commit log配合完成的。
1、Consume Queue
consume queue是消息的逻辑队列相当于字典的目录用来指定消息在物理文件commit log上的位置。
我们可以在配置中指定consumequeue与commitlog存储的目录 每个topic下的每个queue都有一个对应的consumequeue文件比如
${rocketmq.home}/store/consumequeue/${topicName}/${queueId}/${fileName}Consume Queue文件组织如图所示 Consume Queue文件组织示意图
根据topic和queueId来组织文件图中TopicA有两个队列0,1那么TopicA和QueueId0组成一个ConsumeQueueTopicA和QueueId1组成另一个ConsumeQueue。按照消费端的GroupName来分组重试队列如果消费端消费失败消息将被发往重试队列中比如图中的%RETRY%ConsumerGroupA。按照消费端的GroupName来分组死信队列如果消费端消费失败并重试指定次数后仍然失败则发往死信队列比如图中的%DLQ%ConsumerGroupA。死信队列Dead Letter Queue一般用于存放由于某种原因无法传递的消息比如处理失败或者已经过期的消息。 Consume Queue中存储单元是一个20字节定长的二进制数据顺序写顺序读如下图所示 consumequeue文件存储单元格式
CommitLog Offset是指这条消息在Commit Log文件中的实际偏移量Size存储中消息的大小Message Tag HashCode存储消息的Tag的哈希值主要用于订阅时消息过滤订阅时如果指定了Tag会根据HashCode来快速查找到订阅的消息
2、Commit Log
CommitLog消息存放的物理文件每台broker上的commitlog被本机所有的queue共享不做任何区分。 文件的默认位置如下仍然可通过配置文件修改
${user.home} \store\${commitlog}\${fileName}CommitLog的消息存储单元长度不固定文件顺序写随机读。消息的存储结构如下表所示按照编号顺序以及编号对应的内容依次存储。 Commit Log存储单元结构图
3、消息存储实现
消息存储实现比较复杂也值得大家深入了解后面会单独成文来分析(目前正在收集素材)这小节只以代码说明一下具体的流程。
// Set the storage time
msg.setStoreTimestamp(System.currentTimeMillis());
// Set the message body BODY CRC (consider the most appropriate setting
msg.setBodyCRC(UtilAll.crc32(msg.getBody()));
StoreStatsService storeStatsService this.defaultMessageStore.getStoreStatsService();
synchronized (this) {long beginLockTimestamp this.defaultMessageStore.getSystemClock().now();// Here settings are stored timestamp, in order to ensure an orderly globalmsg.setStoreTimestamp(beginLockTimestamp);// MapedFile操作物理文件在内存中的映射以及将内存数据持久化到物理文件中MapedFile mapedFile this.mapedFileQueue.getLastMapedFile();// 将Message追加到文件commitlogresult mapedFile.appendMessage(msg, this.appendMessageCallback);switch (result.getStatus()) {case PUT_OK:break;case END_OF_FILE:// Create a new file, re-write the messagemapedFile this.mapedFileQueue.getLastMapedFile();result mapedFile.appendMessage(msg, this.appendMessageCallback);break;DispatchRequest dispatchRequest new DispatchRequest(topic,// 1queueId,// 2result.getWroteOffset(),// 3result.getWroteBytes(),// 4tagsCode,// 5msg.getStoreTimestamp(),// 6result.getLogicsOffset(),// 7msg.getKeys(),// 8/*** Transaction*/msg.getSysFlag(),// 9msg.getPreparedTransactionOffset());// 10// 1.分发消息位置到ConsumeQueue// 2.分发到IndexService建立索引this.defaultMessageStore.putDispatchRequest(dispatchRequest);
}4、消息的索引文件
如果一个消息包含key值的话会使用IndexFile存储消息索引文件的内容结构如图 消息索引 索引文件主要用于根据key来查询消息的流程主要是
根据查询的 key 的 hashcode%slotNum 得到具体的槽的位置(slotNum 是一个索引文件里面包含的最大槽的数目例如图中所示 slotNum5000000)根据 slotValue(slot 位置对应的值)查找到索引项列表的最后一项(倒序排列,slotValue 总是指向最新的一个索引项)遍历索引项列表返回查询时间范围内的结果集(默认一次最大返回的 32 条记录)
六、消息订阅
RocketMQ消息订阅有两种模式一种是Push模式即MQServer主动向消费端推送另外一种是Pull模式即消费端在需要时主动到MQServer拉取。但在具体实现时Push和Pull模式都是采用消费端主动拉取的方式。
首先看下消费端的负载均衡 消费端负载均衡 消费端会通过RebalanceService线程10秒钟做一次基于topic下的所有队列负载
遍历Consumer下的所有topic然后根据topic订阅所有的消息获取同一topic和Consumer Group下的所有Consumer然后根据具体的分配策略来分配消费队列分配的策略包含平均分配、消费端配置等
如同上图所示如果有 5 个队列2 个 consumer那么第一个 Consumer 消费 3 个队列第二 consumer 消费 2 个队列。这里采用的就是平均分配策略它类似于分页的过程TOPIC下面的所有queue就是记录Consumer的个数就相当于总的页数那么每页有多少条记录就类似于某个Consumer会消费哪些队列。
通过这样的策略来达到大体上的平均消费这样的设计也可以很方面的水平扩展Consumer来提高消费能力。
消费端的Push模式是通过长轮询的模式来实现的就如同下图 Push模式示意图 Consumer端每隔一段时间主动向broker发送拉消息请求broker在收到Pull请求后如果有消息就立即返回数据Consumer端收到返回的消息后再回调消费者设置的Listener方法。如果broker在收到Pull请求时消息队列里没有数据broker端会阻塞请求直到有数据传递或超时才返回。
当然Consumer端是通过一个线程将阻塞队列LinkedBlockingQueuePullRequest中的PullRequest发送到broker拉取消息以防止Consumer一致被阻塞。而Broker端在接收到Consumer的PullRequest时如果发现没有消息就会把PullRequest扔到ConcurrentHashMap中缓存起来。broker在启动时会启动一个线程不停的从ConcurrentHashMap取出PullRequest检查直到有数据返回。
七、RocketMQ的其他特性
前面的6个特性都是基本上都是点到为止想要深入了解还需要大家多多查看源码多多在实际中运用。当然除了已经提到的特性外RocketMQ还支持
定时消息消息的刷盘策略主动同步策略同步双写、异步复制海量消息堆积能力高效通信.......
其中涉及到的很多设计思路和解决方法都值得我们深入研究
消息的存储设计既要满足海量消息的堆积能力又要满足极快的查询效率还要保证写入的效率。高效的通信组件设计高吞吐量毫秒级的消息投递能力都离不开高效的通信。.......
RocketMQ最佳实践
一、Producer最佳实践
1、一个应用尽可能用一个 Topic消息子类型用 tags 来标识tags 可以由应用自由设置。只有发送消息设置了tags消费方在订阅消息时才可以利用 tags 在 broker 做消息过滤。 2、每个消息在业务层面的唯一标识码要设置到 keys 字段方便将来定位消息丢失问题。由于是哈希索引请务必保证 key 尽可能唯一这样可以避免潜在的哈希冲突。 3、消息发送成功或者失败要打印消息日志务必要打印 sendresult 和 key 字段。 4、对于消息不可丢失应用务必要有消息重发机制。例如消息发送失败存储到数据库能有定时程序尝试重发或者人工触发重发。 5、某些应用如果不关注消息是否发送成功请直接使用sendOneWay方法发送消息。
二、Consumer最佳实践
1、消费过程要做到幂等即消费端去重 2、尽量使用批量方式消费方式可以很大程度上提高消费吞吐量。 3、优化每条消息消费过程
三、其他配置
线上应该关闭autoCreateTopicEnable即在配置文件中将其设置为false。
RocketMQ在发送消息时会首先获取路由信息。如果是新的消息由于MQServer上面还没有创建对应的Topic这个时候如果上面的配置打开的话会返回默认TOPIC的RocketMQ会在每台broker上面创建名为TBW102的TOPIC路由信息然后Producer会选择一台Broker发送消息选中的broker在存储消息时发现消息的topic还没有创建就会自动创建topic。后果就是以后所有该TOPIC的消息都将发送到这台broker上达不到负载均衡的目的。
所以基于目前RocketMQ的设计建议关闭自动创建TOPIC的功能然后根据消息量的大小手动创建TOPIC。
RocketMQ设计相关
RocketMQ的设计假定 每台PC机器都可能宕机不可服务 任意集群都有可能处理能力不足 最坏的情况一定会发生 内网环境需要低延迟来提供最佳用户体验 RocketMQ的关键设计 分布式集群化 强数据安全 海量数据堆积 毫秒级投递延迟推拉模式 这是RocketMQ在设计时的假定前提以及需要到达的效果。我想这些假定适用于所有的系统设计。随着我们系统的服务的增多每位开发者都要注意自己的程序是否存在单点故障如果挂了应该怎么恢复、能不能很好的水平扩展、对外的接口是否足够高效、自己管理的数据是否足够安全...... 多多规范自己的设计才能开发出高效健壮的程序。
参考资料
RocketMQ用户指南RocketMQ原理简介RocketMQ最佳实践阿里分布式开放消息服务(ONS)原理与实践2阿里分布式开放消息服务(ONS)原理与实践3RocketMQ原理解析转自https://www.jianshu.com/p/453c6e7ff81c
