地址：https://www.tuicool.com/articles/i2UnYbY

最近在公司做openresty+flume+kafka的前端日誌採集，在測試flume時發現向kafka傳輸的頻寬吞吐才20MB/s，遠遠無法滿足需求。

找到瓶頸

不可盲目

盲目優化是很浪費時間的，一開始只能調調flume引數，改改batchSize之類的引數，結果無功而返。

最終無奈，決定靜心看一下flume的架構原理。

瞭解架構

flume的流水線上，包括source採集磁碟日誌，channel快取採集的日誌，sink將日誌發往遠端，按道理我需要分析出到底哪個環節拖慢了整體的流水線頻寬。

對於流水線系統來說，靠後的環節慢，則會導致之前的環節全部變慢，因為流水線擁塞了。

比如sink處理慢，就導致channel填滿，channel填滿就導致source停止採集。

懷疑的點

source只是簡單的TAILDIR模式，採集目錄下的增量日誌，順序讀磁碟的頻寬遠不止20MB/s，所以我對這個環節的懷疑是最輕的。

channel我採用了memory channel，因為source和sink各只有1個執行緒，鎖競爭不至於成為瓶頸。

sink採用了kafka sink，採用阻塞模型，傳送完一批才會從channel取下一批。

kafka叢集是否存在效能問題呢？為了驗證這一點，我配置了2個channel，讓source採用repliacating模式複製2份流量，併為2個channel各配置1個kafka sink，發現頻寬就是40MB/s了，說明kafka不是問題。

資料說話

效能調優不能靠猜，最好有資料為證，啟動flume時可以配置開啟metrics埠，這是一個http介面，可以查詢實時flume效能指標。

啟動時指定引數：

-Dflume.monitoring.type=http -Dflume.monitoring.port=34545

檢視實時指標：

curl localhost:34545/metrics

返回JSON如下：

{
 "SINK.k1": {
 "ConnectionCreatedCount": "0",
 "BatchCompleteCount": "0",
 "BatchEmptyCount": "257",
 "EventDrainAttemptCount"
 
: "0",
 "StartTime": "1519711797903",
 "BatchUnderflowCount": "3",
 "ConnectionFailedCount": "0",
 "ConnectionClosedCount": "0",
 "Type": "SINK",
 "RollbackCount": "0",
 "EventDrainSuccessCount": "40713702",
 "KafkaEventSendTimer": "1545475",
 "StopTime": "0"
 },
 "SINK.k2": {
 "ConnectionCreatedCount": "0",
 "BatchCompleteCount": "0",
 "BatchEmptyCount": "256",
 "EventDrainAttemptCount": "0",
 "StartTime": "1519711799103",
 "BatchUnderflowCount": "3",
 "ConnectionFailedCount": "0",
 "ConnectionClosedCount": "0",
 "Type": "SINK",
 "RollbackCount": "0",
 "EventDrainSuccessCount": "40713702",
 "KafkaEventSendTimer": "1555840",
 "StopTime": "0"
 },
 "SINK.k3": {
 "ConnectionCreatedCount": "0",
 "BatchCompleteCount": "0",
 "BatchEmptyCount": "256",
 "EventDrainAttemptCount": "0",
 "StartTime": "1519711799335",
 "BatchUnderflowCount": "3",
 "ConnectionFailedCount": "0",
 "ConnectionClosedCount": "0",
 "Type": "SINK",
 "RollbackCount": "0",
 "EventDrainSuccessCount": "40713702",
 "KafkaEventSendTimer": "1556911",
 "StopTime": "0"
 },
 "CHANNEL.c3": {
 "ChannelCapacity": "1000000",
 "ChannelFillPercentage": "0.0",
 "Type": "CHANNEL",
 "EventTakeSuccessCount": "40713702",
 "ChannelSize": "0",
 "EventTakeAttemptCount": "40713962",
 "StartTime": "1519711796033",
 "EventPutAttemptCount": "40713702",
 "EventPutSuccessCount": "40713702",
 "StopTime": "0"
 },
 "CHANNEL.c2": {
 "ChannelCapacity": "1000000",
 "ChannelFillPercentage": "0.0",
 "Type": "CHANNEL",
 "EventTakeSuccessCount": "40713702",
 "ChannelSize": "0",
 "EventTakeAttemptCount": "40713962",
 "StartTime": "1519711796033",
 "EventPutAttemptCount": "40713702",
 "EventPutSuccessCount": "40713702",
 "StopTime": "0"
 },
 "CHANNEL.c1": {
 "ChannelCapacity": "1000000",
 "ChannelFillPercentage": "0.0",
 "Type": "CHANNEL",
 "EventTakeSuccessCount": "40713702",
 "ChannelSize": "0",
 "EventTakeAttemptCount": "40713963",
 "StartTime": "1519711796033",
 "EventPutAttemptCount": "40713702",
 "EventPutSuccessCount": "40713702",
 "StopTime": "0"
 },
 "SOURCE.src_taildir": {
 "EventReceivedCount": "122141106",
 "AppendBatchAcceptedCount": "1566",
 "Type": "SOURCE",
 "AppendReceivedCount": "0",
 "EventAcceptedCount": "122141106",
 "StartTime": "1519711796541",
 "AppendAcceptedCount": "0",
 "OpenConnectionCount": "0",
 "AppendBatchReceivedCount": "1566",
 "StopTime": "0"
 }
}

主要觀察channel的填充率ChannelFillPercentage，如果接近100%說明佇列無法及時被消費，瓶頸在sink端。（上述json是我優化後的，發現channel基本為空）。

優化

既然瓶頸是同步阻塞推送kafka的sink端，那麼顯然增加sink的數量就可以加快channel消費。

一開始我未經瞭解，直接使用了sink gourp為channel配置了2個kafka sink，採用load balance來分發流量，結果發現仍舊20MB/s，毫無提升。

經過思考得知，sink group仍舊採用單執行緒工作，只是充當了2個kafka sink的代理而已，日誌輪轉的被交給2個kafka sink物件，即2個kafka sink物件在同一個執行緒裡交替被呼叫，根本沒有並行能力。

照著這個優化方向，我期望可以配置多個channel，每個channel一個sink執行緒，並讓source將流量均勻的派發給2個channel，從而實現多執行緒併發。

/**
 * 背景: Kafka Sink 單執行緒同步呼叫，吞吐無法繼續提升。
 *
 * 解決方案：自定義實現channel selector, 將source的流量均勻分發到多個channel, 並讓每個channel由一個獨立的kafka sink消費
 *
 * 配置：為source指定selector.type=org.apache.flume.channel.RRChannelSelector
 *
 */
public class RRChannelSelector extends AbstractChannelSelector {
  private static final List<Channel> EMPTY_LIST = new ArrayList<>();
 
  private int rrIndex = 0;
 
  @Override
  public List<Channel> getRequiredChannels(Event event) {
    List<Channel> allChannels = getAllChannels();
 
    int index = rrIndex;
    rrIndex = (rrIndex + 1) % allChannels.size();
 
    List<Channel> result = new ArrayList<>();
    result.add(allChannels.get(index));
    return result;
  }
 
  @Override
  public List<Channel> getOptionalChannels(Event event) {
    return RRChannelSelector.EMPTY_LIST;
  }
 
  @Override
  public void configure(Context context) {
 
  }
}

flume-ng-round-robin-channel-selector

用途

flume-ng預設source -> channel -> sink的流水線配置，會受到sink處理速率的影響，吞吐無法線性提升。

本外掛通過實現自定義channel selector，實現了source均勻派發流量到多個channel，從而可以為每個channel配備一個獨立的sink（執行緒），從而實現吞吐線性提升。

編譯方法

• 將RRChannelSelector.java檔案拷貝到原始碼子路徑：flume-ng-core/src/main/java/org/apache/flume/channel
• 回到flume原始碼根目錄，編譯整個專案：mvn clean install -DskipTests
• 拷貝flume-ng-core/target/flume-ng-core-1.8.0.jar（其中1.8.0是你flume版本）到線上flume環境的lib目錄下覆蓋對應檔案

配置方法

為source指定selector.type，例如：

agent.sources.src_taildir.selector.type = org.apache.flume.channel.RRChannelSelector

為source指定多個channel，例如：

agent.sources.src_taildir.channels = c1 c2 c3

為每個channel指定一個sink，例如：

agent.sinks.k1.channel = c1 agent.sinks.k2.channel = c2 agent.sinks.k3.channel = c3

我們實現一個自定義的Channel Selector類，然後重新編譯flume生成新的jar包，覆蓋到flume即可。

下面是我最終的配置，我配置了3個channel，3個kafka sink，1個source，並配置source的channel selector為我的外掛，從而可以將流量輪轉的發給每一個channel：

# describe the agent
agent_zdm1.sources=src_taildir
agent_zdm1.sinks=k1 k2 k3
agent_zdm1.channels=c1 c2 c3
 
# Describe/configure the source
agent_zdm1.sources.src_taildir.type = TAILDIR
agent_zdm1.sources.src_taildir.positionFile = /root/log-analyze/taildir/taildir_position.json
agent_zdm1.sources.src_taildir.filegroups = f1
agent_zdm1.sources.src_taildir.filegroups.f1 = /data/logs/collect/.*
agent_zdm1.sources.src_taildir.batchSize = 100000
agent_zdm1.sources.src_taildir.selector.type = org.apache.flume.channel.RRChannelSelector
 
#agent_zdm1.sources.src_taildir.backoffSleepIncrement = 2
#agent_zdm1.sources.src_taildir.maxBackoffSleep = 10
 
# Describe the sink
agent_zdm1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
agent_zdm1.sinks.k1.kafka.topic = analytics-zcollect
agent_zdm1.sinks.k1.kafka.bootstrap.servers = localhost:9092
agent_zdm1.sinks.k1.kafka.flumeBatchSize = 1000
agent_zdm1.sinks.k1.kafka.producer.acks = 0
agent_zdm1.sinks.k1.kafka.producer.linger.ms = 100
agent_zdm1.sinks.k1.kafka.producer.batch.size = 100000
 
agent_zdm1.sinks.k2.type = org.apache.flume.sink.kafka.KafkaSink
agent_zdm1.sinks.k2.kafka.topic = analytics-zcollect
agent_zdm1.sinks.k2.kafka.bootstrap.servers = localhost:9092
agent_zdm1.sinks.k2.kafka.flumeBatchSize = 1000
agent_zdm1.sinks.k2.kafka.producer.acks = 0
agent_zdm1.sinks.k2.kafka.producer.linger.ms = 500
agent_zdm1.sinks.k2.kafka.producer.batch.size = 100000
 
agent_zdm1.sinks.k3.type = org.apache.flume.sink.kafka.KafkaSink
agent_zdm1.sinks.k3.kafka.topic = analytics-zcollect
agent_zdm1.sinks.k3.kafka.bootstrap.servers = localhost:9092
agent_zdm1.sinks.k3.kafka.flumeBatchSize = 1000
agent_zdm1.sinks.k3.kafka.producer.acks = 0
agent_zdm1.sinks.k3.kafka.producer.linger.ms = 500
agent_zdm1.sinks.k3.kafka.producer.batch.size = 100000
 
# Use a channel which buffers events in memory
agent_zdm1.channels.c1.type = memory
agent_zdm1.channels.c1.capacity = 1000000
agent_zdm1.channels.c1.transactionCapacity = 100000
 
agent_zdm1.channels.c2.type = memory
agent_zdm1.channels.c2.capacity = 1000000
agent_zdm1.channels.c2.transactionCapacity = 100000
 
agent_zdm1.channels.c3.type = memory
agent_zdm1.channels.c3.capacity = 1000000
agent_zdm1.channels.c3.transactionCapacity = 100000
 
# Bind the source and sink to the channel
agent_zdm1.sources.src_taildir.channels = c1 c2 c3
 
agent_zdm1.sinks.k1.channel = c1
agent_zdm1.sinks.k2.channel = c2
agent_zdm1.sinks.k3.channel = c3

batchSize等引數當然具有一定的意義，但是僅用於優化單個pipeline（流水線），要實現線性擴充套件是需要線上程擴充套件性方面做上述優化工作的。

成果

經過優化，在4核的伺服器上執行4個openresty+1個flume程序，仍舊可以跑出40MB/s的網絡卡外出流量，日誌採集無延遲，達到了我預期中的效果。