35_尚硅谷_Flume高级_自定义Source（代码实现）

好，我们来关注一下，就是我说的这个start跟这个stop啊，等会啊，我看它不一样，我们把它干什么。复制一下对吧，看TRLC复制过来，然后在这个里边我们去来一个class，这个class呢是SOC source包价来一个my source可以吧，SOCMY好继承抽象的这个S，然后呢，啊这个。接口哦，要飞车。呃，没个依赖，没有依赖吗？这个依赖包出了问题吗？这个东西有。呃，然后他呢，他有他也应该有，哎，这个又好又有啊走一下啊，三个奥回车实现相应的方法对吧？走你看一下这里面呢，刚刚才我们看到的内容有一点不一样啊。Start跟stop没了吧，对吧，Start跟stop没了，他这里面呢，主要就是con还有那个什么。来看一下这两个方法，这两个方法呢，对于我们当年没有用啊，一个是be of sleep的一个rement。

增加了一个什么值，每次对吧，还有一个sleep，最这个初始值就是相当于。啊，初始值这两个东西呢，我们就不做推理了，直接就是这样啊，直接这样，我们最核心的就是我们要写的，我们把这个方法拿上。啊，放在一块这个方法呢。给它放在这啊，这两方面呢，暂不做处理，因为跟对比相关的啊，咱们做处理好这个里边我们要配一些东西。啊，一些东西，然后在process里边要写我们最核心的方法。最核心的方法，OK，那这里面呢，要做三件事。第。接收。数据这块括号过一下这个循环。训完造数据。啊，自己在里头造数据，我们就不写那个什么solid通信啊，或者说I物流这些东西去读取数据了，好第二个接触完这数据之后要封装。

为事件。啊，封装为事件，第三步将事件要干什么？传给China对吧，将事件传给。拆了。啊，我们直接写一下China。看China就做这个事儿啊，就这个事返回这状态，接不接着要接收啊，继续啊，要继续传的一个问题，那我们就写这个内容，然后我们需要什么配置文件，再往上面去添加啊去添加好那这块来一下。啊，这块来讲，那这里面呢，按照这个步骤来呗，第一步接收数据，直接解复循环对吧？啊接收数据。啊，接着说句负循环，我那负哎。一呃，每一次呢，我让他发五条数据可以吧，啊，每一次每调用一次发条，调用一次发条OK吧，做这个事啊发五条，这五条数据呢，就是你自己造一下啊，我们之前想的不是你一个even的对吧，这里边事件。

对吧，那有一个even的，但是我们发现它是一个什么。接口是吧，接口，所以我们应该找一下它的一个实现类，在这CTRLH。这就有什么杰还有一个simple，那我们用那个OK吧，用这个，那这个呢，有一个simple。先把一分的到家回正啊，拿到一个一个的实践。啊，拿到一个实践，这个实践里边五条数据，来我们这样写一下啊第二种。构建实践对象。对象。因为我们是自己造的数据，正常的你应该是从外部接收出去，然后把它封装成事件，当然你封装事件前提是有时间这个对象嘛，然后是不是set had set什么。

Body啊，Set head body，那这个head呢，我们也跟那个官方一样，就不要这个head是不是可以不set，我们重点的set什么。Body啊body OK，那第二个第三个事件给事件。设置。值啊，也就是body啊，也是body OK，那这个里面呢，是even的点set body，它里边要一个字节数组。对吧，C数组，那我们就这样，呃，我们就是输入这个I吧，就用I啊这个I呢，我想啊，不不能单独的输入I，我想从外面给他传一个前缀进来。就是这个前面呢，还不在里面写死，我要从配置文件里面给他还进来。我要做这个事，首先呢，你最终是不是应该什么东西加上一个I点什么get best对吧？啊，因为它里面要直接数组嘛，Get best那前缀这个东西我用一个字符串。

对吧。啊，那这个东西呢，我希望他从。配置文件里的，你给我传下来。而且呢，是没有默认值的那种。我不要默认，然后。这块呢，我们加一个，因为里面字符传对吧，这后面呢是一个数字，我加两个杠给它分开，同时我还要一个什么东西呢，后缀。对吧。这个呢，注意我要的这个东西叫前缀，这个叫后缀。它呢，没有默认值。它呢有默认值。

啊，你得从外面给我传进来。哦，我想做这个需求，那我们要在哪个方法里边去做这个事。是不是应该在configuration这里面。而且你看啊，在这里面获取的值在这能用，所以说这两个参数应该定义成全局的对不对。OK，那我们定义。第一全局的全局得了是吧。全局啊的啊，不能还还不能打全局的，他就全局得了，好定全局的那个前缀对吧，And后缀。好的。好，那另外private。类型的一个什么。啊，做初始化，不做初始化，我无所谓往那一放，对吧，那还有一个private。

Six对吧。就做这个事。好，做这事好，那接下来给他们俩去干什么呢？这款叫读取。配置信息。前后缀。负值，负值还记得怎么读的吗？刚才看到里边。Context点。Get。来，你看一下考屁，你看一下的视频是不是有两个方法，一个只传K。一个船K还有一个default value，也就是说如果通过K我没找到，就是你没配对吧，就没找到嘛，你的配置文件里面没有，那我就用什么默认值，哎，如果像第二种你没配这个值就是什么。就是空的了，就没有没有值了，就这意思啊，两种呢，要这样的一个方式来配好，那我们第一个我们说了前缀是没有默认值的，没有默认值应该用一个什么。

K直接单独的一个K吧，好第二个CTRLD复制一份是不是应该是有默认值的对吧？啊是后缀，我们想给它来一个默认值，那在这再来一个什么。默认值，那默认值我们加一个什么东西呢，后面。就是每一条数据后面加一个。呃，加一个爱的硅谷吧，对吧？啊，加一个爱的硅谷就默认的加一个爱的硅谷，好，接下来我们要定义这个什么。定义这个K。那这个K应该怎么定义呢？如果定义完了，它跟什么相关呢？是这样的，你这里边K叫什么名字随便，你叫AA，这个叫BBBB。可以，但是你最终在写配置文件的时候，你要注意你前面a2.3个点K2点什么。AAA。能听懂啊，跟这个是它是这个地方的一个内容。啊，然后如果说你BB那个A2点c.K2点什么。

Bbb等于什么东西要匹配上就行了，什么什么名字无所谓，能听懂啊，那这个呢，为了好看一点，我把它改成什么和。这样你是不是在人家一看配置文件就知道这个是干什么事去的。对吧，因为就像这个，你一看到这个东西，诶，他就pass，你是不是就知道它是可以录进来呀，你给一个AA，人家看到你这个东西，人家知道你这个干什么。对吧，所以呢，你这个配置文件的一个key啊，要介于注意啊，我们就按照这种方式来做。按照这种方式来做好，然后继续这一块，是不是两个东西就等于获取到了对吧？获取到了假如说我们没写prefix，那东西没配它就是空的呀，那他就以杠杠开头的。对吧，啊，这个刚刚开头是这个意思，然后点get，但他要的是字节数组，OK吧？啊，要的是字节数组，好，那接下来我们继续这个东西，有了之后按照官方文档的一个描述，接下来去干什。

是不是将那个东西传给China对吧？好，那get pro，诶这个地方很奇怪的点在于它怎么直接get chinapro，没有方法调。那他怎么知道调的是哪个？有有绑定的对吧？啊，首先呢，负类里边有这个方法吧，啊，分类里边有这个方法，而且写了相应的一个什么实现了啊，写了实现了OK，那这个呢，将世界转给。China这一步是不是刚才我们看到叫直接get China process.process诶你看一下它这个地方还可以干什么。P处理P处理吧，啊，那这个很简单，因为我们看到蓝截器里面是不是也有一个的批次的啊，所以两个都有，那既然我们这样写了，那我们就一个一个的去传一下吧，哎，Even的给它传过去啊，Even的给它传过去，好，然后我们进来看一下这个里边。

点。首先你看一下他做了什么事。这个是不是我们将这个China传给了China process呀，因为我们刚才写的是get China process.process就处理这个事件对吧，那我们一点进来，他先走谁啊。拦截气量嘛，哎，拦截气量。对吧，获得整个这个系列去干什么。拦截这个事件是不是做处理，然后判断这个是不是为空，因为你拦截器里边是不是有可能我们说把数据过滤掉为空啊为空这个方法还是有必要往下走吗。直接它直接为他的吧，好，继续，假如说不为空往下走，这一块才算什么事。是不是我们当时讲的是不是将数据传给，然后他走拦截器，然后走的这边是么。

选择器吧，因为经过拦截器之后，我才能确定选择器里边走的谁了，好，那接下来他经过了谁之后，因为你有可能配的是多个呀。In replicating，是不是这个事件快要传到多个里边，它一匹配上，是不是返回了一个list？因为你有多个都要传，他就去遍历这个例子，好，然后对于每一个China是不是都去发同样的这一个事件。没问题吧？这个代码应该能看懂吧，好继续往下走，你看这个他。获取China里边的一个事物了吧，好，那接下来他这块这个事物开启了，对吧，这个在干什么。Put吧，我们之前说过，是不是这前面有一个put数吗？对吧？好put去提交，如果提交过程当中发生异常，他要做什么事。微。对吧，这一套都有了，而且你看这一块的事物需要我们自己去写，人家写好的，因为我们之前说过，我们关心的是这个数据怎么来，因为你关心的是个source。

你要告诉框架，将来你这个造S读哪一块的数据你就够了，然后呢，你要写给他系统一个内容，你不能说到这一块就结束了，到这块结束了，不好意思，系统说我处理不了，因为你没有传给我China这边的一个处理器，对吧？啊这个呢叫China处理器，在这个里边是我们所说的经过了燃气进量。对吧，然后经过了China选择器，因为经过China选择器里边我们知道有一种叫relic，是有可能当前这个事件要传给多个什么China的，所以你能在这个里边看到的内容是什么。一个复数完那个这个不是一个例子吗。一个集合吧，啊，因为他一去获取发现当前这个事件啊好了，你背了三个China，这三个China呢都要，那我就只能获取完一个一个的发法，那为了每一个保证完成，是不是每一个里面都有一个自己的事务吧，对吧？啊事物开启是这个是核心业务吧，把它放到什么类子里边去提交，提交失败是不回滚，因为异常嘛，对吧？啊异常去做回滚啊，就做回滚好，假如说没有异常，这个finally里是不是必须执行的事物，如果不为空把事物干什么。

关闭掉，因为事五一关闭当前这个事情就已经结束了，证明这个数据真正的传到了哪个里边。拆到里面了对吧，是我完成了啊，保证数据不丢失的，能听懂啊，是这做这个事的，OK，那我们代码其实上没有写到这块啊，因为我们这里面按照人家这个逻辑，人家返回是status对吧，所以呢，我们应该对这个进行一个try catch，如果说它正常的话返回。Ready对吧，也就是这一块来我们在这圈page，那这个里面怎么做，Ctrl out加T。加T快捷键有可以选择包裹的内容，你可以if包裹if else well do well等等这些东西东西可以去包裹，那我们用什么？

拆开包裹一下吧，啊，拆开包裹一下好，Catch里边如果抓到异条了，它是怎么写的？是不是状态等于这个，那既然状态等于这个，说明它前面第一位是什么。状态对吧，最后呢，把这个状态给他一个返回啊，给他做了一个返回，那我们也这样来写一下OK吧，来嗯，这个里面按照他这个想法，第一步是不是定义一个当前的状态。等于那么声明对吧。The等于no。啊，等于一个呢，好，如果这个事件正常完成，这个C等于什么。叫点ready。啊点ready啊ready OK，好，接下来它这个地方。怎么做的？它是我们按跟它一样对吧，Status等于点。Back off对吧，Off，那最后呢，把这个东西干什么？

返回下。啊，最后呢，做一个返回OK了啊，这样呢，就是对于异常我们也做了一个处理，那这里面呢，就去发送一些数据啊，就是发送一个数据，但是你现在这样写了不好，因为这个等会你一启动的时候，它跟疯了一样。五条五条，五条五条跟疯了一样，那这里边呢，我们最好给他加一个什么。睡眠。点sleep对吧，嗯，是个两秒吧，这个有一常是吧，哎，我怎么写的不。而且有异常的，嗯，抓一下。抓一下，把这个异常给他打印下啊，给他打印下，最后呢，回这个当前的一个状态啊，返回当前的一个状态，其实这个东西最好写在哪，因为他前面抓一条没抓到，这边又出一套。

放在里面也可以，但是呢，放在外面是不是也可以，因为我当前这个异常跟你没有什么关系吗？你这个异常是在传输过程当中出现了问题了啊，那这个呢，是最后你都传输完了吧，是不是已经传输完了出来了，然后发生疫苗，那我们单独打印也可以吧，啊这个无所谓，这个呢返回结果。就OK，这个是我们整个的一个编码啊，整个的编码我们再从头来捋一下，这两个东西不要。从头捋一下来看一下，第一个我们在这个地方呢，定义了两个内容。哎，这个有问题吧，没人提醒我。Prefix这个没写。对吧，那我获取出来值没人接收，那有什么意义呢？对吧，少写的这个内容吧，对吧，我前面做了声明，是不是这里面都做赋值啊，这边说了给前后这副值啊，那直接这样写，那不白写了，还是空吗。对吧，你这里面配没配都是空对吧，都是空，OK，那我们就这个来看一下，首先呢，我们定义了两个全局的变量，一个一个对吧前后缀，然后在通过这个con这个方法给这两个东西进行什么赋值，而且呢，我们在设计的时候，这两个东西呢，这两个参数还不一样，一个是没有默认值的，一个是什么。

有默认值的对吧？好，那接下来继续往后走，就是我们最核心的就是这个process方法。这个方法里面呢，你要做三件事情，第一个就是接收数据。只不过说我们当前这个地方用什么。自己造的数据。对吧，你可以从MY等等其他的一个地方去读数据都可以啊，读进来之后把它封装成什么。事件，然后通过get China process.process这个even把这个数据传给。China拦截器就拦截器，然后再传给China选择器，然后让他自己把这个事物正确的完成了，啊是做这个事的，我们核心关心的是这些业务，OK，那我们同样的，呃，因为我们为了避免啊，他跟发疯似的。

我们加了一个什么sleep啊，让他二二千啊，就两秒钟呢去执行一次，也就是说等会呢，启动之后两秒钟打印股条数据，两秒钟打印五条数据，就这样的一个结果能听懂。