flume channel选择器介绍

一：channel选择器

Channel选择器用来解决source接收数据后写入到哪些channel，channel选择器的配置是通过channel处理器完成的，通过在配置文件中使用seletor后缀设置，flume内置了2种channel选择器，一种是复制，一种是分发，默认情况下，如果不特别指定，则默认是复制的形式，在设置选择器的时候，我们也可以设置哪些是必须，哪些是可选的，通过optional来进行设置。

二：复制channel选择器

? 在设置type的时候，type的值为replication，就代表复制，它有一个参数optional,这个参数是可选的，对于optional指定的channel，如果写入数据失败，Source不会抛出异常，相反，对于必须的channel，一旦channle内存满了，或者其它原因导致不能写入，Source将会抛出异常，然后重试。

??新建一个文件，配置如下：

a3.sources = r1
a3.sinks = k1 k2 k3
a3.channels = c1 c2 c3
a3.sources.r1.type=spooldir
a3.sources.r1.spoolDir = /flume/apache-flume-1.6.0-bin/replicat
a3.sources.r1.fileHeader = true
a3.sources.r1.channels = c1 c2 c3
a3.sources.r1.selector.optional = c3
a3.sources.r1.selector.type=replicating
a3.sinks.k1.type = avro
a3.sinks.k1.channel = c1
a3.sinks.k1.hostname = ip
a3.sinks.k1.port = 4444

a3.sinks.k2.type = avro
a3.sinks.k2.channel = c2
a3.sinks.k2.hostname = ip
a3.sinks.k2.port = 5555

a3.sinks.k3.type =avro
a3.sinks.k3.channel  = c3
a3.sinks.k3.hostname = ip
a3.sinks.k3.port = 6666
 
a3.channels.c1.type = memory
a3.channels.c1.capacity=1000
a3.channels.c1.transcationCapacity=100

a3.channels.c2.type = memory
a3.channels.c2.capacity=1000
a3.channels.c2.transcationCapacity=100
 
a3.channels.c3.type = memory
a3.channels.c3.capacity = 1000
a3.channels.c3.transcationCapacity=100

? 同一个数据源经不同的channel，写入不同的sink，其中c3是可选的，意思是如果写入c3失败，该失败会被无视，当我们往监控目录放入文件后，agent通过复制的方式，将内容分发到3个不同的channel，供sink读取，效果如下：

三：分发选择器

分发选择器我们可以理解为，可以根据不同的报文头值进行不同的路由，指定哪些事件可以进入哪些channel。

配置参数如下：

Type:multiplexing

Header:用来检查该事件路由到那个channel

Mapping.*?:?报文头映射的列表

Optional?:?可选的参数，如果写入失败，不会报错

新建一个文件，配置如下:

a0.sources = r1
a0.sinks = k1 k2 k3 k4
a0.channels = c1 c2 c3 c4
a0.sources.r1.type=exec
a0.sources.r1.command= cat  /flume/apache-flume-1.6.0-bin/testFlumeDir/exec.log
a0.sources.r1.host = ip
a0.sources.r1.port = 8888
a0.sources.r1.channels = c1 c2 c3 c4
a0.sources.r1.interceptors = i1
a0.sources.r1.interceptors.i1.type = static
a0.sources.r1.interceptors.i1.key = airline
a0.sources.r1.interceptors.i1.value = CA
a0.sources.r1.selector.type=multiplexing

# airline的值CA,CZ
a0.sources.r1.selector.header = airline
a0.sources.r1.selector.mapping.CA = c1 c2
a0.sources.r1.selector.mapping.CZ = c2
a0.sources.r1.selector.optional.CA = c3
a0.sources.r1.selector.optional.CZ = c4
a0.sources.r1.selector.mapping.default = c4
a0.sinks.k1.type = avro
a0.sinks.k1.channel = c1
a0.sinks.k1.hostname = ip
a0.sinks.k1.port = 4444
 
a0.sinks.k2.type = avro
a0.sinks.k2.channel = c2
a0.sinks.k2.hostname = ip
a0.sinks.k2.port = 5555

a0.sinks.k3.type = avro
a0.sinks.k3.channel = c3
a0.sinks.k3.hostname = ip
a0.sinks.k3.port = 6666

a0.sinks.k4.type = avro
a0.sinks.k4.channel = c4
a0.sinks.k4.hostname = ip
a3.sinks.k4.port = 7777
 
a0.channels.c1.type = memory
a0.channels.c1.capacity=1000
a0.channels.c1.transcationCapacity=100

a0.channels.c2.type = memory
a0.channels.c2.capacity=1000
a0.channels.c2.transcationCapacity=100
 
a0.channels.c3.type = memory
a0.channels.c3.capacity=1000
a0.channels.c3.transcationCapacity=100
 
a0.channels.c4.type = memory
a0.channels.c4.capacity=1000
a0.channels.c4.transcationCapacity=10

配置中通过静态拦截器自定义报文头值，对于报文头里面key为airline,value为CA的会写入到C1，C2，C3这3个channel，且C3是可选的，即使写入C3失败，Source也不会抛出异常，不会重试。效果如下:

C1:

C2:

C3:

C4没收到数据: