flume Interceptor和selector

冷不防 2022-01-16 11:49 322阅读 0赞

[2019独角兽企业重金招聘Python工程师标准>>> ][2019_Python_] ![hot3.png][]

拦截器作用：拦截器是简单的插件式组件，设置在source和channel之间。source接收到的事件，在写入channel之前，拦截器都可以进行转换或者删除这些事件。每个拦截器只处理同一个source接收到的事件。可以自定义拦截器。

常用的拦截器：

1.  时间戳拦截器
    
    flume中一个最经常使用的拦截器 ，该拦截器的作用是将时间戳插入到flume的事件报头中。如果不使用任何拦截器，flume接受到的只有message。时间戳拦截器的配置。

source连接到时间戳拦截器的配置：

[?][Link 1]

<table> 
 <tbody> 
  <tr> 
   <td> 
    <div>
      1 
    </div> 
    <div>
      2 
    </div> 
    <div>
      3 
    </div></td> 
   <td> 
    <div> 
     <code>a1.sources.r1.interceptors&nbsp;=&nbsp;timestamp</code> 
    </div> 
    <div> 
     <code>a1.sources.r1.interceptors.timestamp.type=timestamp</code> 
    </div> 
    <div> 
     <code>a1.sources.r1.interceptors.timestamp.preserveExisting=</code> 
     <code>false</code> 
    </div></td> 
  </tr> 
 </tbody> 
</table>

2. 主机拦截器

主机拦截器插入服务器的ip地址或者主机名，agent将这些内容插入到事件的报头中。时间报头中的key使用hostHeader配置，默认是host。主机拦截器的配置

<table> 
 <tbody> 
  <tr> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">参数</td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">默认值</td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">描述</td> 
  </tr> 
  <tr> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">type</td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;"></td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">类型名称host</td> 
  </tr> 
  <tr> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">hostHeader</td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">host</td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">事件投的key</td> 
  </tr> 
  <tr> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">useIP</td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">true</td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">如果设置为false，host键插入主机名</td> 
  </tr> 
  <tr> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">preserveExisting</td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">false</td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">如果设置为true，若事件中报头已经存在，不会替换host报头的值</td> 
  </tr> 
 </tbody> 
</table>

source连接到主机拦截器的配置：

[?][Link 1]

<table> 
 <tbody> 
  <tr> 
   <td> 
    <div>
      1 
    </div> 
    <div>
      2 
    </div> 
    <div>
      3 
    </div> 
    <div>
      4 
    </div></td> 
   <td> 
    <div> 
     <code>a1.sources.r1.interceptors&nbsp;=&nbsp;host</code> 
    </div> 
    <div> 
     <code>a1.sources.r1.interceptors.host.type=host</code> 
    </div> 
    <div> 
     <code>a1.sources.r1.interceptors.host.useIP=</code> 
     <code>false</code> 
    </div> 
    <div> 
     <code>a1.sources.r1.interceptors.timestamp.preserveExisting=</code> 
     <code>true</code> 
    </div></td> 
  </tr> 
 </tbody> 
</table>

3. 静态拦截器

静态拦截器的作用是将k/v插入到事件的报头中。配置如下

<table> 
 <tbody> 
  <tr> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">参数</td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">默认值</td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">描述</td> 
  </tr> 
  <tr> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">type<br></td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;"></td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">类型名称static</td> 
  </tr> 
  <tr> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">key</td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">key</td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">事件头的key</td> 
  </tr> 
  <tr> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">value</td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">value</td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">key对应的value值</td> 
  </tr> 
  <tr> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">preserveExisting</td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">true</td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">如果设置为true，若事件中报头已经存在该key，不会替换value的值</td> 
  </tr> 
 </tbody> 
</table>

source连接到静态拦截器的配置：

[?][Link 1]

<table> 
 <tbody> 
  <tr> 
   <td> 
    <div>
      1 
    </div> 
    <div>
      2 
    </div> 
    <div>
      3 
    </div> 
    <div>
      4 
    </div> 
    <div>
      5 
    </div></td> 
   <td> 
    <div> 
     <code>a1.sources.r1.interceptors&nbsp;=&nbsp;</code> 
     <code>static</code> 
    </div> 
    <div> 
     <code>a1.sources.r1.interceptors.</code> 
     <code>static</code> 
     <code>.type=</code> 
     <code>static</code> 
    </div> 
    <div> 
     <code>a1.sources.r1.interceptors.</code> 
     <code>static</code> 
     <code>.key=logs</code> 
    </div> 
    <div> 
     <code>a1.sources.r1.interceptors.</code> 
     <code>static</code> 
     <code>.value=logFlume</code> 
    </div> 
    <div> 
     <code>a1.sources.r1.interceptors.</code> 
     <code>static</code> 
     <code>.preserveExisting=</code> 
     <code>false</code> 
    </div></td> 
  </tr> 
 </tbody> 
</table>

4. 正则过滤拦截器

在日志采集的时候，可能有一些数据是我们不需要的，这样添加过滤拦截器，可以过滤掉不需要的日志，也可以根据需要收集满足正则条件的日志。

<table> 
 <tbody> 
  <tr> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">参数</td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">默认值</td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">描述</td> 
  </tr> 
  <tr> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">type<br></td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;"></td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">类型名称REGEX_FILTER</td> 
  </tr> 
  <tr> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">regex</td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">.*</td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;"><span style="font-size:13px;font-family:'Microsoft YaHei UI', 'Microsoft YaHei', SimSun, 'Segoe UI', 'Lucida Grande', Verdana, Arial, Helvetica, sans-serif;color:rgb(42,42,42);line-height:18px;">匹配除“\n”之外的任何个字符</span></td> 
  </tr> 
  <tr> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">excludeEvents</td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">false<br></td> 
   <td style="border-top:rgb(248,248,248) 1px solid;border-right:rgb(248,248,248) 1px solid;border-bottom:rgb(248,248,248) 1px solid;border-left:rgb(248,248,248) 1px solid;">默认收集匹配到的事件。如果为true，则会删除匹配到的event，收集未匹配到的。</td> 
  </tr> 
 </tbody> 
</table>

source连接到正则过滤拦截器的配置：

[?][Link 1]

<table> 
 <tbody> 
  <tr> 
   <td> 
    <div>
      1 
    </div> 
    <div>
      2 
    </div> 
    <div>
      3 
    </div> 
    <div>
      4 
    </div></td> 
   <td> 
    <div> 
     <code>a1.sources.r1.interceptors&nbsp;=&nbsp;regex</code> 
    </div> 
    <div> 
     <code>a1.sources.r1.interceptors.regex.type=REGEX_FILTER</code> 
    </div> 
    <div> 
     <code>a1.sources.r1.interceptors.regex.regex=(rm)|(kill)</code> 
    </div> 
    <div> 
     <code>a1.sources.r1.interceptors.regex.excludeEvents=</code> 
     <code>false</code> 
    </div></td> 
  </tr> 
 </tbody> 
</table>

这样配置的拦截器就只会接收日志消息中带有rm 或者kill的日志。

selector作用：同一个数据源分发到不同的目的地

官网中selector共有两种类型：

Replicating Channel Selector (default)

Multiplexing Channel Selector

![163917\_tCAZ\_2000675.jpg][163917_tCAZ_2000675.jpg]

这两种selector的区别是:Replicating 会将source过来的events发往所有channel,而Multiplexing 可以选择该发往哪些channel。对于上面的例子来说，如果采用Replicating ,那么demo和demo2的日志会同时发往channel1和channel2,这显然是和需求不符的，需求只是让demo的日志发往channel1,而demo2的日志发往channel2。

综上所述，我们选择Multiplexing Channel Selector。这里我们有遇到一个棘手的问题，Multiplexing 需要判断header里指定key的值来决定分发到某个具体的channel，我们现在demo和demo2同时运行在同一个服务器上，如果在不同的服务器上运行，我们可以在 source1上加上一个 host 拦截器（上面有介绍），这样可以通过header中的host来判断event该分发给哪个channel，而这里是在同一个服务器上，由host是区分不出来日志的来源的，我们必须想办法在header中添加一个key来区分日志的来源。

设想一下，如果header中有一个key:flume.client.log4j.logger.source,我们通过设置这个key的值，demo设为app1,demo2设为app2,这样我们就能通过设置：

tier1.sources.source1.channels=channel1 channel2  
tier1.sources.source1.selector.type=multiplexing  
tier1.sources.source1.selector.header=flume.client.log4j.logger.source  
tier1.sources.source1.selector.mapping.app1=channel1  
tier1.sources.source1.selector.mapping.app2=channel2

来将不同项目的的日志输出到不同的channel了。

但是这个header变量从哪里来呢？

解决方法：

1、修改用到的那个source的源码，应用到client端，不同的数据类型添加不同的header值

Event类设计

在Flume中Event是个接口类

**public****interface** Event \{

**public** Map<String, String> getHeaders();

**public****void** setHeaders(Map<String, String> headers);

**public****byte**\[\] getBody();

**public****void** setBody(**byte**\[\] body);

在org.apache.flume.event下, 有两个Event的具体实现类: SimpleEvent, JSonEvent.  
EventBuilder类顾名思义, 采用Builder的方式来组装对象的成员, 并产生最终的对象.

**public****class** EventBuilder \{

**public****static** Event withBody(**byte**\[\] body, Map<String, String> headers) \{

Event event = **new** SimpleEvent();

**if**(body == **null**) \{

body = **new****byte**\[0\];

event.setBody(body);

**if** (headers != **null**) \{

event.setHeaders(**new** HashMap<String, String>(headers));

**return** event;

**public****static** Event withBody(**byte**\[\] body) \{

**return** withBody(body,**null**);

**public****static** Event withBody(String body, Charset charset,

Map<String, String> headers) \{

**return** withBody(body.getBytes(charset), headers);

**public****static** Event withBody(String body, Charset charset) \{

**return** withBody(body, charset, **null**);

2、在source端配置interceptor，通过interceptor在header上设置变量header值

比如：

使用regex\_extractor，对传过来的数据进行处理，提取出type值（如果可以的话，可以在client端的数据格式添加type值，方便使用regex\_extractor提取出来）。

3、在source端自定义interceptor，在interceptor里对处理变量header

Interceptor用于过滤Event，即传入一个Event然后进行过滤加工，然后返回一个新的Event，接口如下：

public interface Interceptor \{  
  
public void initialize();

public Event intercept(Event event);

public List<Event> intercept(List<Event> events);

public void close();  
  
\}

1、public void initialize()运行前的初始化，一般不需要实现（上面的几个都没实现这个方法）；

2、public Event intercept(Event event)处理单个event；

3、public List<Event> intercept(List<Event> events)批量处理event，实际上市循环调用上面的2；

4、public void close()可以做一些清理工作，上面几个也都没有实现这个方法；

5、 public interface Builder extends Configurable 构建Interceptor对象，外部使用这个Builder来获取Interceptor对象。

如果要自己定制，必须要完成上面的2,3,5。

下面，我们来看看org.apache.flume.interceptor.HostInterceptor，其全部代码如下：

/**
     * Licensed to the Apache Software Foundation (ASF) under one
     * or more contributor license agreements.  See the NOTICE file
     * distributed with this work for additional information
     * regarding copyright ownership.  The ASF licenses this file
     * to you under the Apache License, Version 2.0 (the
     * "License"); you may not use this file except in compliance
     * with the License.  You may obtain a copy of the License at
     *
     *     http://www.apache.org/licenses/LICENSE-2.0
     *
     * Unless required by applicable law or agreed to in writing, software
     * distributed under the License is distributed on an "AS IS" BASIS,
     * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
     * See the License for the specific language governing permissions and
     * limitations under the License. */package org.apache.flume.interceptor;import java.net.InetAddress;import java.net.UnknownHostException;import java.util.List;import java.util.Map;import org.apache.flume.Context;import org.apache.flume.Event;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import static org.apache.flume.interceptor.HostInterceptor.Constants.*;/**
     * Simple Interceptor class that sets the host name or IP on all events
     * that are intercepted.<p>
     * The host header is named <code>host</code> and its format is either the FQDN
     * or IP of the host on which this interceptor is run.
     *
     *
     * Properties:<p>
     *
     *   preserveExisting: Whether to preserve an existing value for 'host'
     *                     (default is false)<p>
     *
     *   useIP: Whether to use IP address or fully-qualified hostname for 'host'
     *          header value (default is true)<p>
     *
     *  hostHeader: Specify the key to be used in the event header map for the
     *          host name. (default is "host") <p>
     *
     * Sample config:<p>
     *
     * <code>
     *   agent.sources.r1.channels = c1<p>
     *   agent.sources.r1.type = SEQ<p>
     *   agent.sources.r1.interceptors = i1<p>
     *   agent.sources.r1.interceptors.i1.type = host<p>
     *   agent.sources.r1.interceptors.i1.preserveExisting = true<p>
     *   agent.sources.r1.interceptors.i1.useIP = false<p>
     *   agent.sources.r1.interceptors.i1.hostHeader = hostname<p>
     * </code>
     * */public class HostInterceptor implements Interceptor {
    
      private static final Logger logger = LoggerFactory
              .getLogger(HostInterceptor.class);  private final boolean preserveExisting;  private final String header;  private String host = null;  /**
       * Only {@link HostInterceptor.Builder} can build me   */
      private HostInterceptor(boolean preserveExisting,      boolean useIP, String header) {    this.preserveExisting = preserveExisting;    this.header = header;
        InetAddress addr;    try {
          addr = InetAddress.getLocalHost();      if (useIP) {
            host = addr.getHostAddress();
          } else {
            host = addr.getCanonicalHostName();
          }
        } catch (UnknownHostException e) {
          logger.warn("Could not get local host address. Exception follows.", e);
        }
    
    
      }  @Override
      public void initialize() {    // no-op  }  /**
       * Modifies events in-place.   */
      @Override
      public Event intercept(Event event) {
        Map<String, String> headers = event.getHeaders();    if (preserveExisting && headers.containsKey(header)) {      return event;
        }    if(host != null) {
          headers.put(header, host);
        }    return event;
      }  /**
       * Delegates to {@link #intercept(Event)} in a loop.
       * @param events
       * @return
       */
      @Override
      public List<Event> intercept(List<Event> events) {    for (Event event : events) {
          intercept(event);
        }    return events;
      }  @Override
      public void close() {    // no-op  }  /**
       * Builder which builds new instances of the HostInterceptor.   */
      public static class Builder implements Interceptor.Builder {
    
        private boolean preserveExisting = PRESERVE_DFLT;    private boolean useIP = USE_IP_DFLT;    private String header = HOST;    @Override
        public Interceptor build() {      return new HostInterceptor(preserveExisting, useIP, header);
        }    @Override
        public void configure(Context context) {
          preserveExisting = context.getBoolean(PRESERVE, PRESERVE_DFLT);
          useIP = context.getBoolean(USE_IP, USE_IP_DFLT);
          header = context.getString(HOST_HEADER, HOST);
        }
    
      }  public static class Constants {
        public static String HOST = "host";    public static String PRESERVE = "preserveExisting";    public static boolean PRESERVE_DFLT = false;    public static String USE_IP = "useIP";    public static boolean USE_IP_DFLT = true;    public static String HOST_HEADER = "hostHeader";
      }
    
    }

Constants类是参数类及默认的一些参数：

Builder类是构造HostInterceptor对象的，它会首先通过configure(Context context)方法获取配置文件中interceptor的参数，然后方法build()用来返回一个HostInterceptor对象：

1、preserveExisting表示如果event的header中包含有本interceptor指定的header，是否要保留这个header,true则保留；

2、useIP表示是否使用本机IP地址作为header的value，true则使用IP，默认是true;

3、header是event的headers的key，默认是host。

HostInterceptor：

1、构造函数除了赋值外，还有就是根据useIP获取IP或者hostname；

2、intercept(Event event)方法是设置event的header的地方，首先是获取headers对象，然后如果同时满足preserveExisting==true并且headers.containsKey(header)就直接返回event，否则设置headers:headers.put(header, host)。

3、intercept(List<Event> events)方法是循环调用上述2的方法。

显然其他几个Interceptor也就类似这样。在配置文件中配置source的interceptor时，如果是自己定制的interceptor，则需要对type参数赋值：完整类名+￥Builder,比如com.MyInterceptor$Builder即可。

这样设置好headers后，就可以在后续的流转中通过selector实现细分存储。

转载于:https://my.oschina.net/u/2000675/blog/604202

[2019_Python_]: https://my.oschina.net/u/2663968/blog/3061697
[hot3.png]: /images/20220114/64f804c2350a4c7ebbc557e30b1bff89.png
[Link 1]: http://my.oschina.net/u/2311010/blog/531241?fromerr=TjLxu2PV#
[163917_tCAZ_2000675.jpg]: /images/20220114/3b67ae24f0574da999d0b603ddddd2fc.png