22_尚硅谷_Kafka_拦截器.avi原创

好，接下来我们看一下这个拦截器啊，拦截器其实一看这个名字就知道它干什么作用的吧，就类似于我们当时在那个提到的作用是一样的，就是将你消息拦下来，然后添加点什么东西，什么做一个校验啊，做个校验等等这些东西啊，啊然后他说的是那个0.10版本才开始引入的啊，主要实现那个客户端啊，对于一个定制化的一个控制逻辑，而且这块拦截器主要是那个produce这边的吧。对，肯定的，那你到了消费者那边，你说在新用一个拦截器没有什么意义，对吧，因为你消费过来的数据，你自己要怎么处理，你都拿到了，你自己想怎么处理怎么处理吧，哎，这是写进去一个方式，OK，然后呢，我们要实现一个接口是producer intercept。啊，Producer的一个什么拦截器，然后它里面核心的方法，第一个看还是初始化那个什么配置文件啊，配置文件那块的一个东西，跟那个当时我们看到的那个。自定义分区那个类里边的一个那个方法是一样的。

第二个。的方法。On send的方法，这个东西这个是核心的，如果你要写拦截器的话，这个是比较核心的一个点，这个干什么事呢？就是说他在你发送数据之前会先调用这个方法。发送数据之前一定在之前，如果是之后还有什么意义吗？拦截器你你都写到卡不卡了，你还拦截下来干嘛呢，对吧，还在之前调用的ALL3啊，调用之前，那你可以看他传进来的数据是什么。Producer recorder，就是我们在producer那边封装的那一行一行的数据吧，还是这样的，OK，那你可以把这个数据拿出来做一个什么类型的一个判断啊，或者说对，对于字段做一个校验啊，他认为不合适直接扔了吧，合适的话接着发送吧，是这样的一个过程，这样的一个过程，OK。然后。返回值。这个里边传的是什么？

这个东西好像在哪见过吧？回调函数里边是不是啊，哎，回调函数里边它也同样的，它发送过去之后啊，它如果说成功函数失败，它会在这个方方法里边，哎，关于这个异常是不是为空，你就知道它到底是成功了还是失败了啊最后还有个close方法啊，就是做一些收尾工作的啊，收尾工作的它里边啊，相应的要实现这些方法，实现这些方法OK，然后我们看一下这个案例，这个案例呢设计的。两个主要是为了呢，让大家知道它这个拦截器啊，也是跟那个弗鲁米一样，形成一个拦截器的一个链啊，拦截器的一个链是这样的，OK，然后这个东西呢，嗯，第一个拦截器。我们正常的发送数据对吧，比如说MESSAGE0到这个，然后最终你应该接收到的应该是原始的数据吧，那我们这个两两个拦截器干什么呢？第一个叫时间拦截器，就是在这个数据前面啊。拼接一个时间戳，拼接一个时间戳，第二个计数的一个拦截器干什么呢？就是说统计一下它成功和失败的一个次数。

就是说你发送了十条，有多少成功的，有多少失败的，就刚才这个事儿，其实这两个拦截器这个任务业务并不复杂，对吧？哎，可以搁一块儿的，但是为了大让大家看到什么呢？拦截器链的一个作用啊，拦截器练它也可以，拦截器链就是你如果公司当中真的要用到了这个拦截器的话。对吧，而且你的业务逻辑很复杂的时候，哎，你要知道啊。它这个东西哎，还可以拦截精练，就是说你把逻辑给它分开，给它分开形成一个拦截精练来操作，那我们写一下这个代码。把灯都关了可好？然后新建一个包。叫intercept in。C。拦截器啊，来一个叫customer啊，应该是两个对吧。叫什么啊，Time intercept，看intercept OK，它要实现什么？

Producer pro producer。赛OK，然后把这个实现它相应的一个方法，相应的方法好，那这个。时间拦截器里边我们要干什么呀？添加一个时间戳吧，那添加时间戳，你想想看在哪个方法里面添加呢。On Sunday吧，哎，在发送数据之前，在发送数据之前，OK，你看啊，它传进来的参数是一个record。返回值呢？还是一个最高等，就是你处理这个结果啊，哎，你要给他返回去啊，你要给他返回去啊，那这个地方我们直接来用一个什么。Producer record啊，它里面要什么东西呢？要一个topic，那给他一个topic。

record.topic对吧，我们从上一个数据里边拿第二个。什么KY6啊K的话。给一下吧。record.k好，还有一个VALUE6对吧，VALUE6这个地方我们要做一下改变了啊，做一下改变了。嗯，这块它是之前是正常的一个value，六现在我们要把它变成什么。但时间戳的那system。点他加上一个什么。什么逗号，或者说中划线类似有一个拼接符吧，然后再加上record.value。了。这样的一个数据，对吧，你就把时间都拼进去了，那关于这个什么返回值这块什么close，我们需要操作吗。不需要的啊，因为这里面我们没有什么其他的一些资源的啊，其他的一些资源的是这样的，那这个第一个时间拦截器就搞定了，第二。

叫count。In。CAD括号。啊，它同样的也是实现。Producer吧，Producer，呃，这地方。我们没有定义泛型不太好啊，我们还是同样的给它定义一个什么。这个KV的一个泛型啊，KV的一个泛型。嗯，这块我们把这个方法给它干掉。或者说把这个东西重新。生成一下啊，它这个泛型给它定义出来，泛型定义出来，这样我们后面操作的时候好操作一点，要不然它是object的类型的，对吧，Object的类型的，那还是用一个producer recorder producer record，它里边要的是record.topic啊record点。

然后。点current minutes，添加上我们刚才那个数据，假如说拼接一个逗号，然后加上那个record.value啊，把这个值添加进去啊，加一个泛型。因为我们到时候写那个生产者的时候，是不是也是KV都是词string类型的呀，啊，所以说在这块把发型提前定义好是无所谓的，那这块我们把翻型也加上。这个地方我们应该写一个string。String string string，然后重写它的一个方法。那这个方法里面我们是要计数对吧，成功和失败，那你想想看，这个东西应该在哪写。叫成功和失败。这个累加。应该通过什么东西来判断它成功和失败，然后才能累加吧。那应该这个累加的方法写在哪啊。是不是应该在这个方法里边，哎，返回就成功失败啊，成功了还是失败了，那我们可以干什么，在前面定义一个。

然后那个success。嗯，月等啊，对他那个类型没给我。然后再来一个E，有问题的吧，Count等于A初始好。啊，这个都有了，那我们是不是通过这个判断，如果说这个什么exception等等于now的话，我们把谁加呀，等等于呢，是发送成功的吧，就success。加掉，然后。这个。Count在这对吧，哎，就是根据它的一个异常，如果它异常有的话，有异常的话，那我们就error的，没有异常的话就是成功的。

好，那我们要在哪打印呢？就说你最终是要打印出来的吧。对吧，那要不然计算了你这这个值JVM关闭了，收回收了。那有什么意义呢？统计的这些东西啊，你是不是要给他打印出来啊，那你说在哪打印呢？你在这个方法里面打印合适吗？每一条就是这个方法，是不是每一条都会调用一次，这个方法是不是也是每一条都调用一次，我们最好找一个什么结束的时候调用一次的方法啊。对吧，所以说只有close是这个功能吧，哎，只有close是这个功能，所以说呢，最后我们来一个sot。哪一个啊发送成功。括号，然后加上一个什么啊，这个success。看，然后再加上调数据可以吧，这样的。

啊，然后把这个复制一份发送失败。然后这个地方是arrow。Count来调数据。调数据对吧，这样就能够做打印了，因为close最后做收尾工作的，他一定是在等你producer这个生产完毕之后才会调用这个方法吧，哎，中间是不可能调用的，不可能调用的OK，那要注意还有一个这个方法。这个要不要动。直接返回。必须要动一下吧，把这个方法再改一下吧，要不然你想想看，我们经过这个抗的这个拦截器之后。那所有的返回值都是闹了吧，哎，所有的返回值都闹了，那不行，那我们应该把这个数据原封不动的给他吧，哎，原封不动的给他，哎，是这样的，好，那两个拦截器就写完了。

啊，两个连接器写完了，接下来我们就在这个生产者这用一下吧，啊，用一下来看一下啊。Customer produce。那分这个分区我们就不要了，逐掉，那它有没有关于什么拦截器的一个。肯定也是在这个属性里面添加的吧，啊在属性里面添加的，那我们找一下。呃，关于拦截器的，或者我们直接搜一下叫什么。来下intercept CE，哎，这有吧，Classes啊，而且是classes，看见没？S可以有多个吧，它是一个什么？是你后面这个值啊，要一个list啊，要跟着一个list OK，那我们先把这个先不管其他的，把这个属性给他拿过来。Customer。点put啊，这里边来的是producer con，点刚才我们看到这个值，OK，它后面要一个。

其实我们知道它用一个list，那我们得干什么，用一个什么list吧。对吧，那这里边传什么东西呢。就是这个泛型这个list里面放的是什么数据呢。它这个地方叫这个什么class对吧，放的其实是这个类名，全类名啊，Class一个全类名，那这个里面写什么，全类名是什么。Stream吧，其实是一个string OK，这个叫那个嗯，就叫list吧，啊就叫list，然后我们把这个什么list点。对OK，然后拿一下它那个什么全列名，还有一个两个。在这。看。Copy，然后把它放进来，然后还有一个。抗。

在这个啊，注意你在这个地方添加的顺序就是拦截器链，它的一个执行顺序。能懂这个意思，你们可以做一个什么测试呢？你这个不是时间吗？你可以干一个，哎，这个是就是来一个什么AB in3，来一个B3ABS在数据前面加AB，应了三的数据前面加B，你最后看谁在前面吧，那谁在前面说明谁被后者为物，对不对？因为你是往前加嘛，对吧，往前加你可以做一个测试啊，这块跟大家提前说一下，就是谁先添加谁先执行啊，谁先添加谁接执行，OK，那这个地方。换一个类的。OK，那这个关于这个什么。拦截器这块啊，搞定了，搞定了之后呢，我们做一个测试啊，啊做一个测试，嗯。做一个测试来，我们起一个什么消费者并卡夫卡，然后。Consumer go go。

逗号102，冒号218118，然后。刚刚我们是second。好，那这样他是开始消费了，嗯，然后我们把我们生产者开一下。走一下。成功发送了十条来。这边是不是时间戳也加上去了呀，哎，时间戳也加上去了，是这样的啊，整个的过程就是拦截器是怎么用的，把它关掉。