背景

之前寫(xiě)過(guò)一篇文章 VictoriaLogs：一款超低占用的 ElasticSearch 替代方案講到了我們使用 Victorialogs 來(lái)存儲(chǔ) Pulsar 消息隊(duì)列的消息 trace 信息。l3s28資訊網(wǎng)——每日最新資訊28at.com

圖片l3s28資訊網(wǎng)——每日最新資訊28at.com

而其中的關(guān)鍵的埋點(diǎn)信息是通過(guò) Pulsar 的 BrokerInterceptor 實(shí)現(xiàn)的，后面就有朋友咨詢這塊代碼是否開(kāi)源，目前是沒(méi)有開(kāi)源的，不過(guò)借此機(jī)會(huì)可以聊聊如何實(shí)現(xiàn)一個(gè) BrokerInterceptor 插件，當(dāng)前還沒(méi)有相關(guān)的介紹文檔。l3s28資訊網(wǎng)——每日最新資訊28at.com

其實(shí)當(dāng)時(shí)我在找 BrokerInterceptor 的相關(guān)資料時(shí)就發(fā)現(xiàn)官方并沒(méi)有提供對(duì)應(yīng)的開(kāi)發(fā)文檔。l3s28資訊網(wǎng)——每日最新資訊28at.com

只有一個(gè) additional servlet的開(kāi)發(fā)文檔，而 BrokerInterceptor 只在 YouTube 上找到了一個(gè)社區(qū)分享的視頻。l3s28資訊網(wǎng)——每日最新資訊28at.com

圖片l3s28資訊網(wǎng)——每日最新資訊28at.com

雖說(shuō)看視頻可以跟著實(shí)現(xiàn)，但總歸是沒(méi)有文檔方便。l3s28資訊網(wǎng)——每日最新資訊28at.com

在這之前還是先講講 BrokerInterceptor 有什么用？l3s28資訊網(wǎng)——每日最新資訊28at.com

圖片l3s28資訊網(wǎng)——每日最新資訊28at.com

其實(shí)從它所提供的接口就能看出，在消息到達(dá) Broker 后的一些關(guān)鍵節(jié)點(diǎn)都提供了相關(guān)的接口，實(shí)現(xiàn)這些接口就能做很多事情了，比如我這里所需要的消息追蹤。l3s28資訊網(wǎng)——每日最新資訊28at.com

創(chuàng)建項(xiàng)目

下面開(kāi)始如何使用 BrokerInterceptor： 首先是創(chuàng)建一個(gè) Maven 項(xiàng)目，然后引入相關(guān)的依賴：l3s28資訊網(wǎng)——每日最新資訊28at.com

<dependency>  <groupId>org.apache.pulsar</groupId>  <artifactId>pulsar-broker</artifactId>  <version>${pulsar.version}</version>  <scope>provided</scope>  </dependency>

實(shí)現(xiàn)接口

然后我們便可以實(shí)現(xiàn) org.apache.pulsar.broker.intercept.BrokerInterceptor 來(lái)完成具體的業(yè)務(wù)了。l3s28資訊網(wǎng)——每日最新資訊28at.com

在我們做消息追蹤的場(chǎng)景下，我們實(shí)現(xiàn)了以下幾個(gè)接口：l3s28資訊網(wǎng)——每日最新資訊28at.com

• messageProduced
• messageDispatched
• messageAcked

以 messageProduced 為例，需要解析出消息ID，然后拼接成一個(gè)字符串寫(xiě)入 Victorialogs 存儲(chǔ)中，其余的兩個(gè)埋點(diǎn)也是類(lèi)似的。l3s28資訊網(wǎng)——每日最新資訊28at.com

@Override  public void messageProduced(ServerCnx cnx, Producer producer, long startTimeNs, long ledgerId, long entryId,                              Topic.PublishContext publishContext) {      String ns = getNs(producer.getTopic().getName());      if (!LogSender.checkNamespace(ns)) {          return;      }    String topic = producer.getTopic().getName();      String partition = getPartition(topic);      String msgId = String.format("%s:%s:%s", ledgerId, entryId, partition);      String s = new Event.Publish(msgId, producer.getClientAddress(), System.currentTimeMillis(),              producer.getProducerName(), topic).toString();      LogSender.send(s);  }

編寫(xiě)項(xiàng)目描述文件

我們需要?jiǎng)?chuàng)建一個(gè)項(xiàng)目描述文件，路徑如下：src/main/resources/META-INF/services/broker_interceptor.yml名字也是固定的，broker 會(huì)在啟動(dòng)的時(shí)候讀取這個(gè)文件，其內(nèi)容如下：l3s28資訊網(wǎng)——每日最新資訊28at.com

name: interceptor-namedescription: descriptioninterceptorClass: com.xx.CustomInterceptor

重點(diǎn)是填寫(xiě)自定義實(shí)現(xiàn)類(lèi)的全限定名。l3s28資訊網(wǎng)——每日最新資訊28at.com

配置打包插件

<build>    <finalName>${project.artifactId}</finalName>    <plugins>      <plugin>        <groupId>org.apache.nifi</groupId>        <artifactId>nifi-nar-maven-plugin</artifactId>        <version>1.2.0</version>        <extensions>true</extensions>        <configuration>          <finalName>${project.artifactId}-${project.version}</finalName>        </configuration>        <executions>          <execution>            <id>default-nar</id>            <phase>package</phase>            <goals>              <goal>nar</goal>            </goals>          </execution>        </executions>      </plugin>    </plugins>  </build>

由于 Broker 識(shí)別的是 nar 包，所以我們需要配置 nar 包插件，之后使用 mvn package 就會(huì)生成出 nar 包。l3s28資訊網(wǎng)——每日最新資訊28at.com

配置 broker.conf

我們還需要在 broker.conf 中配置：l3s28資訊網(wǎng)——每日最新資訊28at.com

brokerInterceptors: "interceptor-name"

也就是剛才配置的插件名稱。l3s28資訊網(wǎng)——每日最新資訊28at.com

不過(guò)需要注意的是，如果你是使用 helm 安裝的 pulsar，在 3.1 版本之前需要手動(dòng)將brokerInterceptors 寫(xiě)入到 broker.conf 中。l3s28資訊網(wǎng)——每日最新資訊28at.com

FROM apachepulsar/pulsar-all:3.0.1  COPY target/interceptor-1.0.1.nar /pulsar/interceptors/  RUN echo "/n" >> /pulsar/conf/broker.conf  RUN echo "brokerInterceptors=" >> /pulsar/conf/broker.conf

不然在最終容器中的 broker.conf 中是讀取不到這個(gè)配置的，導(dǎo)致插件沒(méi)有生效。l3s28資訊網(wǎng)——每日最新資訊28at.com

我們是重新基于官方鏡像打的一個(gè)包含自定義插件的鏡像，最終使用這個(gè)鏡像進(jìn)行部署。l3s28資訊網(wǎng)——每日最新資訊28at.com

https://github.com/apache/pulsar/pull/20719我在這個(gè) PR 中已經(jīng)將配置加入進(jìn)去了，但得在 3.1 之后才能生效；也就是在 3.1 之前都得加上加上這行：l3s28資訊網(wǎng)——每日最新資訊28at.com

RUN echo "/n" >> /pulsar/conf/broker.conf  RUN echo "brokerInterceptors=" >> /pulsar/conf/broker.conf

目前來(lái)看 Pulsar 的 BrokerInterceptor 應(yīng)該使用不多，不然使用 helm 安裝時(shí)是不可能生效的；而且官方文檔也沒(méi)用相關(guān)的描述。l3s28資訊網(wǎng)——每日最新資訊28at.com

本文鏈接：http://www.www897cc.com/showinfo-26-43321-0.html如何編寫(xiě)一個(gè) Pulsar Broker Interceptor 插件

聲明：本網(wǎng)頁(yè)內(nèi)容旨在傳播知識(shí)，若有侵權(quán)等問(wèn)題請(qǐng)及時(shí)與本網(wǎng)聯(lián)系，我們將在第一時(shí)間刪除處理。郵件：2376512515@qq.com

上一篇： Python文件操作：JSON、CSV、TSV、Excel和Pickle文件序列化

下一篇： 深入了解Java的GC原理，掌握J(rèn)VM 性能調(diào)優(yōu)！