Flume架构与源码分析-MemoryChannel事务实现

Flume提供了可靠地日志采集功能,其高可靠是通过事务机制实现的。而对于Channel的事务我们本部分会介绍MemoryChannel和FileChannel的实现。

创新互联公司专注为客户提供全方位的互联网综合服务,包含不限于成都网站设计、成都网站制作、梧州网络推广、小程序定制开发、梧州网络营销、梧州企业策划、梧州品牌公关、搜索引擎seo、人物专访、企业宣传片、企业代运营等,从售前售中售后,我们都将竭诚为您服务,您的肯定,是我们最大的嘉奖;创新互联公司为所有大学生创业者提供梧州建站搭建服务,24小时服务热线:13518219792,官方网址:www.cdcxhl.com

首先我们看下BasicChannelSemantics实现:

Java代码

 
 
  1. public abstract class BasicChannelSemantics extends AbstractChannel {   
  2.   //1、事务使用ThreadLocal存储,保证事务线程安全   
  3.   private ThreadLocal currentTransaction   
  4.       = new ThreadLocal();   
  5.    
  6.   private boolean initialized = false;   
  7.   //2、进行一些初始化工作   
  8.   protected void initialize() {}   
  9.   //3、提供给实现类的创建事务的回调   
  10.   protected abstract BasicTransactionSemantics createTransaction();   
  11.   //4、往Channel放Event,其直接委托给事务的put方法实现   
  12.   @Override   
  13.   public void put(Event event) throws ChannelException {   
  14.     BasicTransactionSemantics transaction = currentTransaction.get();   
  15.     Preconditions.checkState(transaction != null,   
  16.         "No transaction exists for this thread");   
  17.     transaction.put(event);   
  18.   }   
  19.   //5、从Channel获取Event,也是直接委托给事务的take方法实现   
  20.   @Override   
  21.   public Event take() throws ChannelException {   
  22.     BasicTransactionSemantics transaction = currentTransaction.get();   
  23.     Preconditions.checkState(transaction != null,   
  24.         "No transaction exists for this thread");   
  25.     return transaction.take();   
  26.   }   
  27.    
  28.   //6、获取事务,如果本实例没有初始化则先初始化;否则先从ThreadLocal获取事务,如果没有或者关闭了则创建一个并绑定到ThreadLocal。   
  29.   @Override   
  30.   public Transaction getTransaction() {   
  31.    
  32.     if (!initialized) {   
  33.       synchronized (this) {   
  34.         if (!initialized) {   
  35.           initialize();   
  36.           initialized = true;   
  37.         }   
  38.       }   
  39.     }   
  40.    
  41.     BasicTransactionSemantics transaction = currentTransaction.get();   
  42.     if (transaction == null || transaction.getState().equals(   
  43.             BasicTransactionSemantics.State.CLOSED)) {   
  44.       transaction = createTransaction();   
  45.       currentTransaction.set(transaction);   
  46.     }   
  47.     return transaction;   
  48.   }   
  49. }   

MemoryChannel事务实现

首先我们来看下MemoryChannel的实现,其是一个纯内存的Channel实现,整个事务操作都是在内存中完成。首先看下其内存结构:

1、首先由一个Channel Queue用于存储整个Channel的Event数据;

2、每个事务都有一个Take Queue和Put Queue分别用于存储事务相关的取数据和放数据,等事务提交时才完全同步到Channel Queue,或者失败把取数据回滚到Channel Queue。

MemoryChannel时设计时考虑了两个容量:Channel Queue容量和事务容量,而这两个容量涉及到了数量容量和字节数容量。

另外因为多个事务要操作Channel Queue,还要考虑Channel Queue的动态扩容问题,因此MemoryChannel使用了锁来实现;而容量问题则使用了信号量来实现。

在configure方法中进行了一些参数的初始化,如容量、Channel Queue等。首先看下Channel Queue的容量是如何计算的:

Java代码

 
 
  1. try {   
  2.   capacity = context.getInteger("capacity", defaultCapacity);   
  3. } catch(NumberFormatException e) {   
  4.   capacity = defaultCapacity;   
  5. }   
  6.    
  7. if (capacity <= 0) {   
  8.   capacity = defaultCapacity;   
  9. }    

即首先从配置文件读取数量容量,如果没有配置则是默认容量(默认100),而配置的容量小于等于0,则也是默认容量。

接下来是初始化事务数量容量:

Java代码

 
 
  1. try {   
  2.   transCapacity = context.getInteger("transactionCapacity", defaultTransCapacity);   
  3. } catch(NumberFormatException e) {   
  4.   transCapacity = defaultTransCapacity;   
  5. }   
  6. if (transCapacity <= 0) {   
  7.   transCapacity = defaultTransCapacity;   
  8. }   
  9. Preconditions.checkState(transCapacity <= capacity,   
  10. "Transaction Capacity of Memory Channel cannot be higher than " +   
  11.         "the capacity.");   

整个过程和Channel Queue数量容量初始化类似,但是***做了前置条件判断,事务容量必须小于等于Channel Queue容量。

接下来是字节容量限制:

Java代码

 
 
  1. try {   
  2.   byteCapacityBufferPercentage = context.getInteger("byteCapacityBufferPercentage", defaultByteCapacityBufferPercentage);   
  3. } catch(NumberFormatException e) {   
  4.   byteCapacityBufferPercentage = defaultByteCapacityBufferPercentage;   
  5. }   
  6. try {   
  7.   byteCapacity = (int)((context.getLong("byteCapacity", defaultByteCapacity).longValue() * (1 - byteCapacityBufferPercentage * .01 )) /byteCapacitySlotSize);   
  8.   if (byteCapacity < 1) {   
  9.     byteCapacity = Integer.MAX_VALUE;   
  10.   }   
  11. } catch(NumberFormatException e) {   
  12.   byteCapacity = (int)((defaultByteCapacity * (1 - byteCapacityBufferPercentage * .01 )) /byteCapacitySlotSize);   
  13. }    

byteCapacityBufferPercentage:用来确定byteCapacity的一个百分比参数,即我们定义的字节容量和实际事件容量的百分比,因为我们定义的字节容量主要考虑Event body,而忽略Event header,因此需要减去Event header部分的内存占用,可以认为该参数定义了Event header占了实际字节容量的百分比,默认20%;

byteCapacity:首先读取配置文件定义的byteCapacity,如果没有定义,则使用默认defaultByteCapacity,而defaultByteCapacity默认是JVM物理内存的80%(Runtime.getRuntime().maxMemory() * .80);那么实际byteCapacity=定义的byteCapacity * (1- Event header百分比)/ byteCapacitySlotSize;byteCapacitySlotSize默认100,即计算百分比的一个系数。

接下来定义keepAlive参数:

Java代码

 
 
  1. try {   
  2.   keepAlive = context.getInteger("keep-alive", defaultKeepAlive);   
  3. } catch(NumberFormatException e) {   
  4.   keepAlive = defaultKeepAlive;   
  5. }    

keepAlive定义了操作Channel Queue的等待超时事件,默认3s。

接着初始化Channel Queue:

Java代码

 
 
  1. if(queue != null) {   
  2.   try {   
  3.     resizeQueue(capacity);   
  4.   } catch (InterruptedException e) {   
  5.     Thread.currentThread().interrupt();   
  6.   }   
  7. } else {   
  8.   synchronized(queueLock) {   
  9.     queue = new LinkedBlockingDeque(capacity);   
  10.     queueRemaining = new Semaphore(capacity);   
  11.     queueStored = new Semaphore(0);   
  12.   }   
  13. }    

首先如果Channel Queue不为null,表示动态扩容;否则进行Channel Queue的创建。

首先看下***创建Channel Queue,首先使用queueLock锁定,即在操作Channel Queue时都需要锁定,因为之前说过Channel Queue可能动态扩容,然后初始化信号量:Channel Queue剩余容量和向Channel Queue申请存储的容量,用于事务操作中预占Channel Queue容量。

接着是调用resizeQueue动态扩容:

Java代码

 
 
  1. private void resizeQueue(int capacity) throws InterruptedException {   
  2.   int oldCapacity;   
  3.   synchronized(queueLock) { //首先计算扩容前的Channel Queue的容量   
  4.     oldCapacity = queue.size() + queue.remainingCapacity();   
  5.   }   
  6.    
  7.   if(oldCapacity == capacity) {//如果新容量和老容量相等,不需要扩容   
  8.     return;   
  9.   } else if (oldCapacity > capacity) {//如果老容量大于新容量,缩容   
  10.     //首先要预占老容量-新容量的大小,以便缩容容量   
  11. if(!queueRemaining.tryAcquire(oldCapacity - capacity, keepAlive, TimeUnit.SECONDS)) {   
  12.    //如果获取失败,默认是记录日志然后忽略   
  13. } else {   
  14.   //否则,直接缩容,然后复制老Queue的数据,缩容时需要锁定queueLock,因为这一系列操作要线程安全   
  15.       synchronized(queueLock) {   
  16.         LinkedBlockingDeque newQueue = new LinkedBlockingDeque(capacity);   
  17.         newQueue.addAll(queue);   
  18.         queue = newQueue;   
  19.       }   
  20.     }   
  21.   } else {   
  22.     //如果不是缩容,则直接扩容即可   
  23.     synchronized(queueLock) {   
  24.       LinkedBlockingDeque newQueue = new LinkedBlockingDeque(capacity);   
  25.       newQueue.addAll(queue);   
  26.       queue = newQueue;   
  27. }   
  28. //增加/减少Channel Queue的新的容量   
  29.     queueRemaining.release(capacity - oldCapacity);   
  30.   }   
  31. }   
  32.    
  33. 到此,整个Channel Queue相关的数据初始化完毕,接着会调用start方法进行初始化:   
  34. public synchronized void start() {   
  35.   channelCounter.start();   
  36.   channelCounter.setChannelSize(queue.size());   
  37.   channelCounter.setChannelCapacity(Long.valueOf(   
  38.           queue.size() + queue.remainingCapacity()));   
  39.   super.start();   
  40. }    

此处初始化了一个ChannelCounter,是一个计数器,记录如当前队列放入Event数、取出Event数、成功数等。

之前已经分析了大部分Channel会把put和take直接委托给事务去完成,因此接下来看下MemoryTransaction的实现。

首先看下MemoryTransaction的初始化:

Java代码

 
 
  1. private class MemoryTransaction extends BasicTransactionSemantics {   
  2.   private LinkedBlockingDeque takeList;   
  3.   private LinkedBlockingDeque putList;   
  4.   private final ChannelCounter channelCounter;   
  5.   private int putByteCounter = 0;   
  6.   private int takeByteCounter = 0;   
  7.   public MemoryTransaction(int transCapacity, ChannelCounter counter) {   
  8.     putList = new LinkedBlockingDeque(transCapacity);   
  9.     takeList = new LinkedBlockingDeque(transCapacity);   
  10.     channelCounter = counter;   
  11.   }    

可以看出MemoryTransaction涉及到两个事务容量大小定义的队列(链表阻塞队列)、队列字节计数器、另外一个是Channel操作的计数器。

事务中的放入操作如下:

Java代码

 
 
  1. protected void doPut(Event event) throws InterruptedException {   
  2.   //1、增加放入事件计数器   
  3.   channelCounter.incrementEventPutAttemptCount();   
  4.   //2、estimateEventSize计算当前Event body大小   
  5.   int eventByteSize = (int)Math.ceil(estimateEventSize(event)/byteCapacitySlotSize);   
  6.   //3、往事务队列的putList中放入Event,如果满了,则抛异常回滚事务   
  7.   if (!putList.offer(event)) {   
  8.       throw new ChannelException(   
  9.       "Put queue for MemoryTransaction of capacity " +   
  10.         putList.size() + " full, consider committing more frequently, " +   
  11.         "increasing capacity or increasing thread count");   
  12.   }   
  13.   //4、增加放入队列字节数计数器   
  14.   putByteCounter += eventByteSize;   
  15. }    

整个doPut操作相对来说比较简单,就是往事务putList队列放入Event,如果满了则直接抛异常回滚事务;否则放入putList暂存,等事务提交时转移到Channel Queue。另外需要增加放入队列的字节数计数器,以便之后做字节容量限制。

接下来是事务中的取出操作:

Java代码

 
 
  1. protected Event doTake() throws InterruptedException {   
  2.   //1、增加取出事件计数器   
  3.   channelCounter.incrementEventTakeAttemptCount();   
  4.   //2、如果takeList队列没有剩余容量,即当前事务已经消费了***容量的Event   
  5.   if(takeList.remainingCapacity() == 0) {   
  6.     throw new ChannelException("Take list for MemoryTransaction, capacity " +   
  7.         takeList.size() + " full, consider committing more frequently, " +   
  8.         "increasing capacity, or increasing thread count");   
  9.   }   
  10.   //3、queueStored试图获取一个信号量,超时直接返回null   
  11.   if(!queueStored.tryAcquire(keepAlive, TimeUnit.SECONDS)) {   
  12.     return null;   
  13.   }   
  14.   //4、从Channel Queue获取一个Event   
  15.   Event event;   
  16.   synchronized(queueLock) {//对Channel Queue的操作必须加queueLock,因为之前说的动态扩容问题   
  17.     event = queue.poll();   
  18.   }   
  19.   //5、因为信号量的保证,Channel Queue不应该返回null,出现了就不正常了   
  20.   Preconditions.checkNotNull(event, "Queue.poll returned NULL despite semaphore " +   
  21.       "signalling existence of entry");   
  22.   //6、暂存到事务的takeList队列   
  23.   takeList.put(event);   
  24.   //7、计算当前Event body大小并增加取出队列字节数计数器   
  25.   int eventByteSize = (int)Math.ceil(estimateEventSize(event)/byteCapacitySlotSize);   
  26.   takeByteCounter += eventByteSize;   
  27.   return event;   
  28. }   

接下来是提交事务:

Java代码

 
 
  1. protected void doCommit() throws InterruptedException {   
  2.   //1、计算改变的Event数量,即取出数量-放入数量;如果放入的多,那么改变的Event数量将是负数   
  3.   int remainingChange = takeList.size() - putList.size();   
  4.   //2、  如果remainingChange小于0,则需要获取Channel Queue剩余容量的信号量   
  5.   if(remainingChange < 0) {   
  6.     //2.1、首先获取putByteCounter个字节容量信号量,如果失败说明超过字节容量限制了,回滚事务   
  7.     if(!bytesRemaining.tryAcquire(putByteCounter, keepAlive, TimeUnit.SECONDS)) {   
  8.       throw new ChannelException("Cannot commit transaction. Byte capacity " +   
  9.         "allocated to store event body " + byteCapacity * byteCapacitySlotSize +   
  10.         "reached. Please increase heap space/byte capacity allocated to " +   
  11.         "the channel as the sinks may not be keeping up with the sources");   
  12.     }   
  13.     //2.2、获取Channel Queue的-remainingChange个信号量用于放入-remainingChange个Event,如果获取不到,则释放putByteCounter个字节容量信号量,并抛出异常回滚事务   
  14.     if(!queueRemaining.tryAcquire(-remainingChange, keepAlive, TimeUnit.SECONDS)) {   
  15.       bytesRemaining.release(putByteCounter);   
  16.       throw new ChannelFullException("Space for commit to queue couldn't be acquired." +   
  17.           " Sinks are likely not keeping up with sources, or the buffer size is too tight");   
  18.     }   
  19.   }   
  20.   int puts = putList.size();   
  21.   int takes = takeList.size();   
  22.   synchronized(queueLock) {//操作Channel Queue时一定要锁定queueLock   
  23.     if(puts > 0 ) {   
  24.       while(!putList.isEmpty()) { //3.1、如果有Event,则循环放入Channel Queue   
  25.         if(!queue.offer(putList.removeFirst())) {    
  26.           //3.2、如果放入Channel Queue失败了,说明信号量控制出问题了,这种情况不应该发生   
  27.           throw new RuntimeException("Queue add failed, this shouldn't be able to happen");   
  28.         }   
  29.       }   
  30.     }   
  31.     //4、操作成功后,清空putList和takeList队列   
  32.     putList.clear();   
  33.     takeList.clear();   
  34.   }   
  35.   //5.1、释放takeByteCounter个字节容量信号量   
  36.   bytesRemaining.release(takeByteCounter);   
  37.   //5.2、重置字节计数器   
  38.   takeByteCounter = 0;   
  39.   putByteCounter = 0;   
  40.   //5.3、释放puts个queueStored信号量,这样doTake方法就可以获取数据了   
  41.   queueStored.release(puts);   
  42.   //5.4、释放remainingChange个queueRemaining信号量   
  43.   if(remainingChange > 0) {   
  44.     queueRemaining.release(remainingChange);   
  45.   }   
  46.   //6、ChannelCounter一些数据计数   
  47.   if (puts > 0) {   
  48.     channelCounter.addToEventPutSuccessCount(puts);   
  49.   }   
  50.   if (takes > 0) {   
  51.     channelCounter.addToEventTakeSuccessCount(takes);   
  52.   }   
  53.    
  54.   channelCounter.setChannelSize(queue.size());   
  55. }    

此处涉及到两个信号量:

queueStored表示Channel Queue已存储事件容量(已存储的事件数量),队列取出事件时-1,放入事件成功时+N,取出失败时-N,即Channel Queue存储了多少事件。queueStored信号量默认为0。当doTake取出Event时减少一个queueStored信号量,当doCommit提交事务时需要增加putList 队列大小的queueStored信号量,当doRollback回滚事务时需要减少takeList队列大小的queueStored信号量。

queueRemaining表示Channel Queue可存储事件容量(可存储的事件数量),取出事件成功时+N,放入事件成功时-N。queueRemaining信号量默认为Channel Queue容量。其在提交事务时首先通过remainingChange = takeList.size() - putList.size()计算获得需要增加多少变更事件;如果小于0表示放入的事件比取出的多,表示有- remainingChange个事件放入,此时应该减少-queueRemaining信号量;而如果大于0,则表示取出的事件比放入的多,表示有queueRemaining个事件取出,此时应该增加queueRemaining信号量;即消费事件时减少信号量,生产事件时增加信号量。

而bytesRemaining是字节容量信号量,超出容量则回滚事务。

***看下回滚事务:

Java代码

 
 
  1. protected void doRollback() {   
  2.     int takes = takeList.size();   
  3.     synchronized(queueLock) { //操作Channel Queue时一定锁住queueLock   
  4.       //1、前置条件判断,检查是否有足够容量回滚事务   
  5.       Preconditions.checkState(queue.remainingCapacity() >= takeList.size(), "Not enough space in memory channel " +   
  6.           "queue to rollback takes. This should never happen, please report");   
  7.       //2、回滚事务的takeList队列到Channel Queue   
  8.       while(!takeList.isEmpty()) {   
  9.         queue.addFirst(takeList.removeLast());   
  10.       }   
  11.       putList.clear();   
  12.     }   
  13.     //3、释放putByteCounter个bytesRemaining信号量   
  14.     bytesRemaining.release(putByteCounter);   
  15.    
  16.     //4、计数器重置   
  17.     putByteCounter = 0;   
  18.     takeByteCounter = 0;   
  19.     //5、释放takeList队列大小个已存储事件容量   
  20.     queueStored.release(takes);   
  21.     channelCounter.setChannelSize(queue.size());   
  22.   }   
  23. }    

也就是说在回滚时,需要把takeList中暂存的事件回滚到Channel Queue,并回滚queueStored信号量。

【本文是专栏作者张开涛的原创文章,作者微信公众号:开涛的博客,id:kaitao-1234567】

网站栏目:Flume架构与源码分析-MemoryChannel事务实现
文章URL:http://www.mswzjz.com/qtweb/news42/208342.html

网站建设、网络推广公司-创新互联,是专注品牌与效果的网站制作,网络营销seo公司;服务项目有等

广告

声明:本网站发布的内容(图片、视频和文字)以用户投稿、用户转载内容为主,如果涉及侵权请尽快告知,我们将会在第一时间删除。文章观点不代表本网站立场,如需处理请联系客服。电话:028-86922220;邮箱:631063699@qq.com。内容未经允许不得转载,或转载时需注明来源: 创新互联