93.腾讯云EMR-实时数仓搭建-DWS层-交易域-SKU粒度下单需求-关联维表-异步IO-说明原创

好，那。我们之前呢，是已经把这个。旁路缓存都搞定了。对吧，啊，用这个缓存的方式，确实我们看到明显的这个效果了。呃，但是这个里边呢，有一个点啊，我们想一下，我们上午看到了，诶最早的时候呢，我们看到一个数据说先是150。150毫秒。对吧，那个时候呢，我们说一秒钟一条。对吧，一条要一秒钟。因为六六个表的数据嘛，对吧，好，那我们优化到了十秒十毫秒或者说九八毫秒吧。六条啊。然后呢，我们说20条每秒，注意啊，我们说的都是单个平行度对吧，那后面呢，我们又优化到了一毫秒啊，你基本上应该是小于一毫秒。

对吧，那我们认为呢是多少啊，它一条数据要五毫秒，要五毫秒。这是50对吧，这是五毫秒一条数据五毫秒，那也就是说200条每秒又扩大了十倍嘛，对吧，200条每秒好。呃，那其实我们这样去计算啊，它有一个点在里边，什么点呢？就是比如说每条数据它要消耗50毫秒，那两条数据默认直接。乘以二。对吧，那那我在想难道不对吗？就是乘以二啊，对，没问题，那我们想啊，那你觉得这50毫秒，或者说哪怕我最后我们一秒钟或者150毫秒，对吧，它的时间主要花在哪儿了呢。首先啊，你要记住这个点，一条数据50毫秒对吧，两条数据就是100毫秒，这是一个点，第二那我们50毫秒也好，还有150，还有这个一毫秒也好，他这个时间主要是花在哪儿。

主要花在哪，那家说，诶那我知道跟这个数据库的连接。对吧，那有一个问题，连接我都已经建立好了，对吧，连原数据都已经有了，他为什么还要这个一毫秒呢。对吧，他不是直接拿过来呢。啊，那是因为他有这个等待时间对吧，连接你是有了，原数据你也有了，那你呢，是代码在这儿，数据库在这儿对吧，我要发起请求，然后呢，他给你。拿到数据之后去找找你要的数据对吧，然后给你做一个响应，其实我们最重要的时间是不是花在这儿。相当于什么呢？诶，比如说我有多条数据同时到的，对吧，那ABC都来了，但是呢，我A发送请求怎么样，你得。等着。啊，那大家一起等着，你到了也没用对吧，好，那B来了。

那一样的，A出去了嘛，对吧，C早早来了，等等着，这也就是所谓的叫什么呢？叫同步的方式。而我们现在所用的这个map function。它就是一个同步函数。啊，所谓的同步函数指的就是说，那我们知道弗林格它是有自己每一个算子都有自己的缓冲区的，对吧，缓冲区满了，它就会通知这个上游。那你别给我发了，那上游慢慢的满了，最终呢，他会通知到S别接收了，就会产生了这个什么，就是后面产生了这个反压。产生反压呢，导致前面延迟。对吧，它会导致这个前面产生这个延迟，对吧，那这样的一个情况啊好。那既然这样的话。我们就知道了。他数据来了，你不知道操作，因为你要用同步，那我们想着，那你不是就是发请求，然后等着嘛，我们想着那ABC3条数据来了，我能不能三个人都去发这个请求，然后一起等着，这个效率是不是就高了。

能明白吗？所以刚才我们提到了，诶，一条数据来了50毫秒，两条数据直接乘以二，那是因为咱们默认就建立在什么同步的这种想法上去思考的，对吧？默认咱们认为它是同步的，就是第一条数据来了，哪怕你两条数据同时到。真的就是同时到，哪怕两条数同时到，对吧，我也得一条一条的处理，所以你这个时间延迟呢，必须是一个累加，但是。既然你两条来了同时来了，我能不能说两条同时发起请求啊，那我俩一起等着，这样的话效率是不是就提高了？能明白吗？对吧，这是所谓的异步的方式，但是你不能用map函数去实现异步，这个不可能的，因为map函数它本身就是一个同步，就算你有同时有ABC，哪怕你有十条数据一起来了，不好意思，我得一个一个处理，我必须把A处理完了，我才会处理B，才会处理C，所以我们要用异步函数，在fli里边它有一个异步函数，就是只要来了我就直接执行，我不用等第一条执行完。

这个跟map有很大的一个区别了，那如果我们实现这个功能，那就好呀。对吧，因为我们可以多条数据同时怎么样。进行。啊，同时去发送请求，同时等待，然后呢，谁先拿到响应谁先走嘛，对吧，因为这管理微表对不对，谁先拿到了谁走啊然后通知拿到了，哎，就按顺序走就完了。对吧，这个可以提高我们的效率，也就是我们第二种方式用什么呢？异步IO对吧？那这个点呢，我们需要通过这个官方文档给大家做一个介绍啊，在data stream API当中有一个operators，这里边的东西基本上我们就介绍完了啊。

Window学过了吧，Join操作process方式还有一个异步l as不是同步吗？对吧？As异步啊E点开。啊，我看的直接是13这个文档啊，好，那么这块讲的是异步IO的一个必要性，其实讲的就是延迟，你看这个就好了，对吧，这就刚才我们所说的那同步IO的方式，A来了对吧，A发送请求，中间的是等待时间wait。对吧，那A响应好，A走了，A走了之后呢，B才能发起请求，那你看E不呢，你只要来了你就可以发，你不用等。你不用等我A处理完了B才能发对吧，那我呢，大家同时等着，然后呢，拿到响应直接走。对吧，上面是讲的必要性，必要性啊，其实讲的就是中间这个延迟比较高嘛，对吧，比较高啊，大家就比如说啊这边。

呃，这个东西啊。他干什么与当与需要与外部系统进行交互的时候，对吧，他举个例子，他说干什么呢？呃，当我们。使用。With啊，Date store in in the database啊，在存储在数据库里边的数据呢，准备用这个数据啊，来丰富我们流里边事件的时候，那么刚好我们现在这个现象就是对吧？我们要用base里边的数据来丰富我们的招，对吧？呃，One need to tellcare，一定要注意这个什么，That communication。这个交互延迟吧，交互延迟啊，与第三方external外部系统的交互延迟，对吧，往往呢，它会影响到我们整个流的。流应用的整个工作对吧？啊，他说的很清楚啊，那当然这块呢，给你详细介绍了一下，他说什么东西，他说当我们必须要跟这个外部数据库呢，进行交互的数据进行交互的时候，呃，For example对吧，In the map function，在map function里边，这就意味着用的是同步的方，就刚才我们说了map它是同步对吧，它里边没办法实现异步，那也就是说我用多线程啊，不好意思也不行。

你想想看，就算你在map函数里边使用多线程，那我A来了，你开个线程对吧，B来了开个线程，那就是这不可以吗？不好意思，你是两个人，可以独用独立的线程，这个没问题，但是我A已经把这个线程用完了，已经结束了，你B才能拿到你的线程，他俩不是同时拿这个线程工作的，能明白吗？对吧？所以这个是不行的啊，用map就不行，它意味着就是一个同步的方式，对吧？他说什么呢？一个请求被发送到了这个数据库，And my方，对吧，Wait until。等待的直到什么。直到那个响应被接收到的。

直到这个响应被接收到，就是说我们要等待同步的方向，In many cases，在很多的案例当中。多线程也要等啊，什么意思啊，你说的是什么？Map function的吗？你说的是map里边用多线程，你没听懂我的意思是吗？就是我举个例子啊，就是假如说我用map function AB同时来了，你假如说你要用多线程，对吧，这里边可以有很多线程你可以用，比如说A呢，我我我等会儿呢拿这个线程啊，B呢，等会拿这个线。但是这里面有个问题，只要你用的是map function。只要你用的是map方，那么我必须等A处理完，我B才能被处理，能明白吗？就是你，你B想拿现成，你得处理才能拿现成吧，你在麦克防控阵里边能听懂这个意思吗？就是我A拿到现成，我处理完了，你B才能进来去获取你的现场。

我A没处理完的时候，你B连线程都获取不到。能明白吗？所以这个地方哪怕你用map function。多线程都没用。人家web方式就是一个同步的方式，你用多线程也没用，你要求干什么事呢？你必须AB来了，我就能够立马被处理。我立马就要被处理，我才有用啊，我用多线程才有用，对吧？A来了，A立马处理，拿这个线程走了，好，那B来了，拿这个线程走了，但是map方阵里面怎么样？A来了，拿着这个线程不好意思，我呢还没处理完，B你别进来。进不来，门关了，相当于加了一把锁，能，那我这样说能明白吗？或者就相当于线程池上我加了一把锁。线程池上我加了一把锁啊，必须要把线程还回来，我才会给另外一个人现场，我这样去解释能明白吗？

他不调用它B进来，他不做运算，那有什么用啊对吧，多线程当然也要等啊，就是你拿不到你的线程。我A没处理完的时候，你拿不到你的现场。对吧，这个意思好好，那如果用E。L就不一样了，它意味着什么呢？意味着单个并行度的函数实例，对吧？可以处理，处理什么呢？Many requests conculy，主要在这。同时处理多个请求，And receive。The responses。Conculy，对吧，它可以同时接收多个响应，就在这儿，所以呢，所以干什么，这个等待时间can be overlay将会被什么压缩就是。共同的等待时间嘛，就看上去这个等待时间被压缩了，这是我们说的对吧，啊，它这就意味着可以更多的实力呢，拥有更高的流的吞吐量。

对吧，吞吐量就是我们吞吐量会更高，你本来是200条每秒，我现在可能两千两万。对吧，啊，当然这个跟你的并行度啊也有关系，其实啊，对吧，跟你的线程数也有关系啊，当然了这边。其实我们这种方案也也采取了这种方案也也也说了啊，这也是一个优化方案，这个优化方案什么呢？他说你要想提高这个吞吐量啊。如果你你调用的啊是这个map function也可以怎么做呢？To a very high。就是到一个非常高的冰度也可以。对吧，啊，但是呢，这个就消耗资源嘛，大家都知道对吧，那生产环境当中呢，我们肯定如果你数据量特别特别大的时候，你必然要增增加冰度，就算你已经用了缓存，已经用了eboo，你也得怎么样。加机器加。并行度吧，你不可能不加的，那怎么可能啊，对吧啊，但是呢，他也告诉我们呢，但是通常会带来什么一个非常高的资源消耗啊，拥有很多的并度的map方实例，意味着更多的task线程，然后呢，Li内部的网络交互，与外部系统与数据库的什么网络连接，Buff内存等等的，就是资源相应的会怎么样。

提高对吧，这个很好理解，就是说我们假如说不用EIO。不用这个旁路缓存，我就单独的去增加这个并行度，我也可以提高吞吐量，当然没问题啊，对吧，你本来一个并行度，我现在增加了十个，那不也扩展了十倍了，没问题吧，这个点我们就已经聊到过了。对吧，好，这是ebo的一个必要性，对吧？啊必要性就是说跟外部系统交互的时候，这在我们这边对吧？好，那如果你不做优化，它可能会带来什么问题啊。假如说与。外部系统。做交互的时候，我没有做优化，没有做任何优化，它可能会存在什么问题啊，未来。

可能会引发flink的什么问题？刚才其实我提到了它可能会产生什么样的问题。嗯。对，就是这个被压或者叫反压都可以对吧？啊那呃，如果说你的上游是这种主动消费的，比如说卡夫卡，他到任务到不会挂掉或者怎么样的，对吧，他一他会怎么样呢？最后反压主体反馈反馈到SS端，SS就不接收新的数据了。那就会产生。数据积压，卡不卡的数据积压，网络延迟会变得越来越高。对吧，我生产者。发送的数据你没有及时的。

消费掉产生数据积压，对吧，产生这样的一个问题啊。