23__尚硅谷_Kafka_KafkaStream.avi原创

好，接下来我们看一下这个卡普卡stream啊，它本身自己自带的一个流失的一个处理，流失的一个处理，OK，然后他说是一个看源项目的一个组成部分，那卡卡斯肯定是他一个自己的组成部分了，OK，然后看一下它的特点就是特点，特别是这块写的特别扯淡啊，这这个特点是官方文档里边的，官文档里边的什么高扩展弹性融合，然后什么轻量就是它的，这更重要的就是它轻量级。就是因为它太轻了，也没人用，就是说功能太太弱了嘛，太弱了，然后你看他第三个叫完全继承他百分百，你卡不卡自己的这个版本金融，那你不扯淡嘛，是吧，你卡不卡自己的框架，自己一个组成部分你都不兼容，你让人家怎么用啊，对吧，所以说官方文档有时候写写这种话了，写这种话啊，然后什么实时性啊，实时性。啊呃，这块需要说一下的，并非VP处理这块。嗯，它那个卡普卡string啊，它跟stop一样，它就是来一条数据处理一次，来一条数据处理一次，不像Spark，当时你们学Spark的时候，Spark它其实是一个VP次处理的一个框架，就是说它不是一条一条数据来处理的，它会存一下，就是说你处理数据啊，可能是两秒的一个数据，五秒的一个数据，那就这样的，它也是有一个窗口，然后窗口随着移动的啊，所以说Spark呢，那个东西呢，它不是它一个实时性没有STEM高啊，因为它就是因为它是VP次处理的。

啊，VB族处理的，但是它VB族处理啊，所以导致它这个吞吐量特别大，比那个storm要大多了，因为它是一堆一堆数据处理的，对吧，STEM来一条就处理掉，来一条处理掉，当你前面来的数据量，就是说写入的速度过大的时候，那STEM扛不住的。STEM扛不住的啊，所以说现在storm用的相对来说呃在下降，那Spark用的多，那现在还有一个东西，大家应该可能听过叫这个。弗导这个东西啊，也在呈上升趋势，呈上升趋势是这样的啊，整个的过程啊，因为那个Spark它也是V品次处理的，它的一个实时性不是说那么高，不是那么高，他因为他一次处理的什么两秒数据，五秒数据啊，啊是批是一个V批次的一个处理啊，到时候你们学这块的时候啊就知道了啊，知道了OK，然后它是实时性，因为它是跟stop一样，真正的一个流失处理啊，就是说一条数据一条数据来处理的啊，这是它的一个特点，稍微了解一下就行了，然后是这个后面为什么要有这个卡卡stream，然后说现在已经有了很多这个流失处理框架了，什么Spark呀，Storm啊等等啊等等，为什么还要出这个卡卡swim，因为它轻了。

大可以很快的用，所以说它的一个功能性很弱，你看这个什么第一第二第三第四等等，都是说它的一个成本比较低，就是说它这个轻量级啊，都是在说他这个轻量级，但是公司里边呢。他也不在乎这些什么轻量级这些问题，对吧，你功能性太弱了，那就不用，那就不用啊。

好，这是前面一些特点啊，稍微了解一下就行了，这这些东西就觉得都没必要看了，这些文字之前还让大家之前讲过什么架构啊，之后呢，要把那下面文字要读一读嘛，这块东西你就随意啊，你看你想读就读一下吧，不想读就算了。然后我们看一下这个卡不卡stream啊，一个案例，然后我们这个需求做的是什么呢？就是实时的处理这个有这个这三个监控号，就说如果他包括这个三个监控号的，我们把这个三个监控号给他处理掉，给他处理掉就这个意思啊，就做这个事啊，就随便提了一个需求吧，你不是三个监控号，你去一个监控号都无所谓吧，就是给一个业务，给一个业务，OK，然后是这样的，这个东西呢，嗯，卡不卡stream。等会你在创建的时候用出来就知道它传的一个参数啊，是要一个什么build，那个build呢，是是一个拓的一个build啊，是一个拓扑结构，这个整个的拓扑结构呢，就是这三个组件。

你到时候会看到熟悉的think。啊，你能看到这些东西啊，它也一样的，也就是圆筒，它一个托骨结构呢，也是由圆有尘槽，有中间的一个处理过程，处理过程只不过它处理过程不是叫China，因为你这是一个流失处理框架，China仅仅是做一个缓存吧，不做任何处理，对不对啊，哎，所以它叫这个process process。啊，Process，它是process是一个处理的一个进程啊，就是为了处理数据的，所以它跟那个form那块有一点点区别，就是因为form仅仅是做传输数据的，而这个卡卡STEM呢，它还有一定的计算能力啊，所以中间这个组件不一样啊，中间这个组件不一样，那我们整个的是这样的，通过生产者发送数据，这个数据呢，发送给topic first topic first，然后呢，经过中间的这个。处理过程，处理过程之后呢，我们将数据发送给那个second，最后你可以起一个消费者，呃，将那个second里边数据给它消费出来啊，给他消费出来整个的流程，然后我们的一个。

这是我们那个托结构里边的source，这是think，这是中间的process啊，这个东西构成了一个大的一个托结构啊托结构等会我们写代码就就能感觉到啊，等会才能感觉到这整个的一个需求，整个的一个数据流是这样走的，这样走的OK，那同样的我们还是操作一下吧。啊。嗯，再建一个包，把这个东西都关掉。嗯，应该建一个包，或者说直接建一个class吧。呃，叫卡普卡，Streams k fkr。STEM。它同样的也是一个。肯定要有慢方法驱动的，因为是要启动它的，要启动它的，对，那这个里边呢，我们在写这个之前啊，先要搞一个依赖。

搜一下这个依赖吧。哎，这个不是啊，找一下找一会这个。问一下这个卡不卡？呃，用的这个依赖呢，是这个卡卡stems卡STEMOK，我们找到我们自己的版本，我他妈版本是0.11.0.2吧，啊这个。把这个拿过来，CRC。放在那个PM里面。一来看到没导进来。嗯，我这个地方居然被我删掉了，那让他下一下好，咱们网速还可以是吧。那我们用一个什么，同样的也是卡不卡STEM。嗯，我我建的这个名字跟那个官方的一样了是吧。卡不要不要找自己，到时候你们要见那个名字不要这个对吧，搞这个吧，卡卡包下的包下。

嗯，这个就他写的权利名，因为我当前在这个，那我把这个类名还是改一下吧，要不然看着太难受了是吧。然后改成不要这个S。啊，不要这个S，然后我们就这样就OK了吧，把这个包导一下，包导一下这样就好，那么来看一下它里边要的参数。要一个什么，或者说要一个什么，对吧，然后还要一个什么。Technologyology build build这个东西就是我们刚才提到的那个什么，就是拓扑的意思啊，哎，拓扑的整个的一个构建，那我们先写一个一个来用一个什么technology。To Build。来看他要什么东西。不要参数对吧，那我们直接拿它这个返回值看一看，Build build OK，那把这个拿进来。

把拿还要一个什么。什么对吧，我们也给他准备一个用一个什么。哎呀。把这个也给他，先把这两个参数给他，那这样你这个流就就有了吗？这样来卡位值卡不卡，然后最终怎么运行呢？卡卡点start。撒了，那这个东西，我们刚才提到的那个什么。什么first second，还有中间处理过程都还没体现呢，对吧，因为你这个叫to的一个build，但是你有放东西的嘛，你有拓真正的拓扑结构吗？你只是有一个架子嘛，这个东西你只用了这个当前那个对象，你起码要往这个build里边干什么。放点东西吧，哎，这个地方是创建。

托福。对象对象，然后是创建一个配置文件。创建一个配置文件，OK，然后接下来是构建拓结构，构建拓结构那就拿着这个build。点看见没，And source and think and process对吧，就是我们刚才提到的那个source think。那个中间的process处理过程，那从源头开始添加吧，从源头开始添加，添加一个这呃，首先它要一个什么名字对吧，然后要一个topics，要一个topics OK，那我们来。呃，一个名字，那我们就叫S吧，是。好，然后是抛我们还是first。好，这个是原已经添加了吧，再添加什么。

A process添加process OK，这个地方有三个参数，第一个还是它这个名字对吧，第二个是process，那我们能想象到这个里边。这个类啊，我们应该在这个类对象里面写我们自己的一个什么处理过程，对吧？哎，处理过程那最后还有一个parent names。那想一下这个parent name是什么意思呢？对，就是那个S的内，就因为你要构建拓扑结构，就类似于我们在写form那个配置文件，最后那一部分绑定。对吧，哎，绑定，那它在拓扑结构里面，要把这个东西进行一个绑定好，那首先第一个我们来一个什么。Process p。CESSO2啊，这个名字倒无所谓啊，好，第二个我们先放着，第三个我们知道应该写的是。注意这个一定要保持一致啊。逗号多一个吗？不多吧，是这样的，中间因为少一个那个数据对吧，两个数据我们先不管那个。

点爱，还有一个谁啊？Think对吧，Think你看也一样，Name topic parent name对吧？哎，也就是说名字是什么，就是定义，就类似于我们做的一个操作吧，定义声明最后绑定啊三个三个东西OK，然后它这个名字呢，我们就叫think。好，然后第二个topic对吧，我们second。啊叫second，好，第三个parent name，那我们把这个什么中间的处理过程给他，处理过程给他，那接下来我们就搞这个事情，这个他要一个什么。Process来又一个PROCESS6，它是一个。接口啊，又出现了一个接口，那这个接口你就最好不要在这后面写实现方法。

不要在这写实验方法了，因为它里边的一个逻辑啊啊。或者这样，它里面返回的一个是是一个什么。呃，它里面要调用这个，它也是重写了一个，要实现一个方法，对吧，一个方法，但是它返回值是一个process，那在这个地方你又一个什么。吧，那又是一个什么？又是一个接口，那这个东西就显得这个代码就看起来很不舒服了吧，你看而且当前这个接口，这个接口来看一下它的一个方法。还挺多的是吧，哎，这个看起来很不舒服，那我们可以专门。写一个类来实现它吧，来实现它OK，在这。嗯，我们的流呢。我们没写那个包是吧，你们把包写上叫STEMSTEMSTEM。放在这个包里。

Doesn'呃，因为这边有错嘛，等会我们写，先把这些东西都写完吧，呃，我们写一个类，这个类呢，叫刚才提到的log process CEO走一下，然后要实现什么。Process pro so这个接口啊，注意要导一下那个。卡夫卡STEM下的包吧，你不要导错包了，包导第一个啊，好，它里边有一个泛型，那你觉得这个泛型应该怎么定义？瑞能不能行？4040可以，但是比较麻烦。比较麻烦，因为是这样的，为什么比较麻烦呢？你看整个的拓扑结构当中，首先你是不是有一个数据写到那个first里边，对吧，First呢，它对它进行了一个，就是生产者那边进行了一个序列化，序列化然后你整个的过程。

没有经过什么消费者，没有经过消费者，那也就没有反序列化过程，也就是说你这里边传输的数据都是直接输图。都是字节数组，所以说你这个地方如果写成SP的话，你还要自己转一下，又要做那个什么。自己那个反序列化过程吧，就比较麻烦，所以说这块的分型我们定义成什么，就定义成这个代代数组。YT的速度。啊，就定义成这个就够了，因为它中间就是拿半的数组来传的，OK，然后实现它的一个方法。首先哪哪个方法，首先有个什么初始化方法。啊，也是脑袋什么上下文对吧，哎，这个东西第二个process这个东西才是我们你看着。Bed bed的是我们对应的什么KV啊，这块才是我们处理的一个重点，处理一个重点。

OK，那这个地方我们就可以用一个什么。Logo process了啊，你有一个logo process了，是这样的啊，那处理过程它自己就能解决到，它会回调里面的一个process方法，Process方法OK，那这块同样的也可以用拉姆达表达式来表来来替代吧，来看一下这一块的一个拉拉么达表达式啊。空了，什么玩意都没了，对吧，就这样就直接定一条就行了，OK，然后接下来我们写什么核心的一个业务逻辑在这块吧，啊在这块OK，那我们要干什么事啊，刚才说的。是不是要把Y6拿出来？把它那个什么三个进括号给它去掉吧，哎，三个进括号去掉，那首先你要去进括号的话，你是不是要把它转成spring啊。对吧，又一个。String把这个BAT二传进去啊，这是一行一行的数据，一行一行数据拿过来啊，这就是获取一行数据。

好，第二个获取到银行数据之后，就是真正的什么去除。那个。这玩意吧，去除这玩意，那怎么来来点。然后。三个监控号把它替换成什么，什么都不要吧，哎，替换成什么都不要就替换掉就OK了，替换掉就OK了，之后你是不是要把这个东西给他。转成半的数组才能写出去啊。对吧，哎，转成半的数组才能写出去，OK，那这个对转成半的数组。But。还是什么来直接点跟就可以了。

获取一下这个，把它付给这个F号。不行啊，速度不应该这样写，前面就不要声明了，对吧？哎，不给P2，那最终我们是要干什么写出去吧。麻烦了，这这在这怎么写啊，你用一个创建一个IO流写出去。之前是不是看到什么用上下文啊，来看一下这个上下文里边有什么什么什么方法点。Forward forward就是写出去的一个方法吧，哎，往外推嘛，Forward。它里面也放了一个什么KV对吧，放了一个KV。那好，那这个我知道是调用forward的方法来写出去，那怎么在这，你是你起码这个方法在这儿调用吧。对吧。在这调用呢，那你应该把它拿出来，是不是你应该在外面干什么。

对这个东西进行一个声明是不是。就是说拿其他方法里边。属性的话，是不是你可以在外面把它声明成什么成员变量嘛，啊成员变量context。然后这个地方怎么写。等于context。OK，那你提升了成员变量以后，你在这。context.for what，然后K的话是这个best我们不动吧，K还是自己，然后value呢？拜是那个二拜这个。那这样就能写出去了啊，就能写出去了。这个地方，这个地方我们应该把它自己把它接收一下吧，接收一下，要不然听完之后还是自己没变吧，你之后是不是又把赖转化出去了了，应该接收一下这个地方。

好，那这个就是我们核心的一个业务逻辑啊，一个业务逻辑，然后我们测一下。嗯。他说什么？Missing required con。ID啊，这个地方要指定一个ID在这，呃，Publish点。因为这个东西写的少，我忘了。他报错了，就想起来了，给他一个ID啊，叫卡夫卡。啊卡对。然后再走一下。

嗯。还有问题。哎。真的。连哪个鸡群都都不给是吧。So。下课了是吧。哈多102报号2181，他一个机群，哎2181了是吧，瞎写的909290922181是那个从K部的一个端口吧。这还有什么纸没设吗？OK了对吧，哎，这就走成功了，它这两个一个是MDID，还有一个它那个连击群肯定要有的吧，啊连机群肯定要有的，OK，你看它是一个什么。阻塞的进程吧，啊一直在运行，那接下来我们来一个测试，呃，首先请一个消费者。三的消费者有了吧，哎，直接起，然后起一个什么。

生产者吧。这个还CTRLC干不掉。CT model卡不卡？B卡不卡？CS of console杠杠。Brook list吧，啊，这几个参数要记得住。杠list啊，刚OK好多102括号90929092杠杠，那你这个写往哪个地方写啊first吧。好，节目号出来了，我先来一个什么，Hello，看他能不能接收到。Hello，没问题吧，哎，能收到，那接下来我来一个什么呢？Hello。我们爱着硅谷，可以吧，走。是不是三个监控号就干掉了呀，啊就是说这个就比如说处理中间所一个业务逻辑啊，当然我们设定的一个业务逻辑呢，比较简单啊，因为它确实也不能做什么复杂的事啊，做不了什么太太复杂的事，可以在这个卡卡这做一个清洗数据啊，清洗数据这块是可以的。