我正在使用 flume-ng 将应用程序生成的数据聚合到一个中央位置进行处理。生产者将事件发布到本地 HTTP 源，然后通过 Avro 接收器转发给收集器。收集器通过 Avro 源接收来自 Avro 接收器的事件。所有生产者都将事件复制到两个收集器。
我注意到一些与我有关的细微差别，并且我无法通过有关 flume-ng 的文档和周围信息完全理解我的配置的完整影响。Apple 对事件发布与生产者的比较表明并非所有事件都到达收集器。成功交付的事件延迟高于预期。偶尔我们会看到事件在收集器的前门晚了几个小时到达。我们还在下游和上游的 Flume-ng 日志中遇到间歇性异常。
- 在生产者上使用文件通道让我相信我们仍然是持久的，flume-ng最终会在上游获取事件（对吗？）。
- 随着收集器利用内存通道，我们是否将自己暴露在数据丢失中？如果是这样，是否有任何技术可以确定丢失了多少事件，或者至少是它们丢失的频率？
- 收集者的交易能力低于生产者，是否会出现问题？
- 需要保证交付的文件通道的合理检查点间隔是多少？我假设间隔是flume-ng确保事件在本地磁盘上的频率（例如，根据我的配置，我预计在断电时最多1s的数据丢失等）。
下面是每个阶段的 flume-ng 配置的相关片段以及堆栈跟踪片段。想知道是否有人可以根据上述问题（假设有）阐明配置中的差异。
上游异常：
(org.apache.avro.ipc.NettyServer$NettyServerAvroHandler.channelClosed:209) - Connection to /<ip address>:<port> disconnected.
(org.apache.avro.ipc.NettyServer$NettyServerAvroHandler.exceptionCaught:201) - Unexpected exception from downstream.
下游异常：
(org.apache.flume.SinkRunner$PollingRunner.run:160) - Unable to deliver event. Exception follows.
org.apache.flume.EventDeliveryException: Failed to send events
at org.apache.flume.sink.AbstractRpcSink.process(AbstractRpcSink.java:382)
at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
at java.lang.Thread.run(Thread.java:679)
Caused by: org.apache.flume.FlumeException: NettyAvroRpcClient { host: <producer ip>, port: <producer port> }: RPC connection error
at org.apache.flume.api.NettyAvroRpcClient.connect(NettyAvroRpcClient.java:161)
at org.apache.flume.api.NettyAvroRpcClient.connect(NettyAvroRpcClient.java:115)
at org.apache.flume.api.NettyAvroRpcClient.configure(NettyAvroRpcClient.java:590)
at org.apache.flume.api.RpcClientFactory.getInstance(RpcClientFactory.java:88)
at org.apache.flume.sink.AvroSink.initializeRpcClient(AvroSink.java:127)
at org.apache.flume.sink.AbstractRpcSink.createConnection(AbstractRpcSink.java:209)
at org.apache.flume.sink.AbstractRpcSink.verifyConnection(AbstractRpcSink.java:269)
at org.apache.flume.sink.AbstractRpcSink.process(AbstractRpcSink.java:339)
... 3 more
Caused by: java.io.IOException: Error connecting to /<producer ip>:<producer port>
at org.apache.avro.ipc.NettyTransceiver.getChannel(NettyTransceiver.java:261)
at org.apache.avro.ipc.NettyTransceiver.<init>(NettyTransceiver.java:203)
at org.apache.avro.ipc.NettyTransceiver.<init>(NettyTransceiver.java:152)
at org.apache.flume.api.NettyAvroRpcClient.connect(NettyAvroRpcClient.java:147)
... 10 more
Caused by: java.net.NoRouteToHostException: No route to host
at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:592)
at org.jboss.netty.channel.socket.nio.NioClientSocketPipelineSink$Boss.connect(NioClientSocketPipelineSink.java:396)
at org.jboss.netty.channel.socket.nio.NioClientSocketPipelineSink$Boss.processSelectedKeys(NioClientSocketPipelineSink.java:358)
at org.jboss.netty.channel.socket.nio.NioClientSocketPipelineSink$Boss.run(NioClientSocketPipelineSink.java:274)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1146)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
... 1 more
生产者配置：
agent.sources.t1d.type = HTTP
agent.sources.t1d.bind = <source ip>
agent.sources.t1d.port = <source port>
agent.sources.t1d.channels = c01 c02 c03
agent.sources.t1d.selector.type = replicating
agent.channels.c01.type = FILE
agent.channels.c01.checkpointDir = /path/to/checkpoint1
agent.channels.c01.dataDirs = /path/to/data1
agent.channels.c01.capacity = 4000000
agent.channels.c01.transactionCapacity = 10000
agent.channels.c01.checkpointInterval = 1000
agent.channels.c01.keep-alive = 3
agent.channels.c02.type = FILE
agent.channels.c02.checkpointDir = /path/to/checkpoint2
agent.channels.c02.dataDirs = /path/to/data2
agent.channels.c02.capacity = 4000000
agent.channels.c02.transactionCapacity = 10000
agent.channels.c02.checkpointInterval = 1000
agent.channels.c02.keep-alive = 3
agent.channels.c03.type = FILE
agent.channels.c03.checkpointDir = /path/to/checkpoint3
agent.channels.c03.dataDirs = /path/to/data3
agent.channels.c03.capacity = 4000000
agent.channels.c03.transactionCapacity = 10000
agent.channels.c03.checkpointInterval = 1000
agent.channels.c03.keep-alive = 3
agent.sinks.s01.type = AVRO
agent.sinks.s01.hostname = <collector ip>
agent.sinks.s01.port = <collector port>
agent.sinks.s01.channel = c01
agent.sinks.s01.batch-size = 1000
agent.sinks.s01.compression-type = deflate
agent.sinks.s01.compression-level = 5
agent.sinks.s02.type = AVRO
agent.sinks.s02.hostname = <collector ip>
agent.sinks.s02.port = <collector port>
agent.sinks.s02.channel = c02
agent.sinks.s02.batch-size = 1000
agent.sinks.s02.compression-type = deflate
agent.sinks.s02.compression-level = 5
agent.sinks.s03.type = FILE_ROLL
agent.sinks.s03.channel = c03
agent.sinks.s03.sink.directory = /path/to/producer/reconciliation
agent.sinks.s03.sink.rollInterval = 1200
采集器A配置：
agent.sources.a.type = avro
agent.sources.a.bind = <source ip>
agent.sources.a.port = <source port>
agent.sources.a.channels = c01 c03
agent.sources.a.compression-type = deflate
agent.channels.c01.type = memory
agent.channels.c01.capacity = 1000000
agent.channels.c01.transactionCapacity = 1000
agent.channels.c03.type = memory
agent.channels.c03.capacity = 1000000
agent.channels.c03.transactionCapacity = 1000
agent.sinks.s01.type = FILE_ROLL
agent.sinks.s01.channel = c01
agent.sinks.s01.sink.directory = /path/to/storage/s01
agent.sinks.s01.sink.rollInterval = 300
agent.sinks.s03.type = FILE_ROLL
agent.sinks.s03.channel = c03
agent.sinks.s03.sink.directory = /path/to/storage/s03
agent.sinks.s03.sink.rollInterval = 300
收集器 B 配置：
agent.sources.b.type = avro
agent.sources.b.bind = <source ip>
agent.sources.b.port = <source port>
agent.sources.b.channels = c11 c13
agent.sources.b.compression-type = deflate
agent.channels.c11.type = memory
agent.channels.c11.capacity = 1000000
agent.channels.c11.transactionCapacity = 1000
agent.channels.c13.type = memory
agent.channels.c13.capacity = 1000000
agent.channels.c13.transactionCapacity = 1000
agent.sinks.s11.type = FILE_ROLL
agent.sinks.s11.channel = c11
agent.sinks.s11.sink.directory = /path/to/storage/s11
agent.sinks.s11.sink.rollInterval = 300
agent.sinks.s13.type = FILE_ROLL
agent.sinks.s13.channel = c13
agent.sinks.s13.sink.directory = /path/to/storage/s13
agent.sinks.s13.sink.rollInterval = 300