1 /**
2 * <p>
3 * Initializes the channel if it is not already, then checks to see
4 * if there is an open transaction for this thread, creating a new
5 * one via <code>createTransaction</code> if not.
6 * @return the current <code>Transaction</code> object for the
7 * calling thread
8 * </p>
9 */
10 @Override
11 public Transaction getTransaction() {
12
13 if (!initialized) {
14 synchronized (this) {
15 if (!initialized) {
16 initialize();
17 initialized = true;
18 }
19 }
20 }
21
22 BasicTransactionSemantics transaction = currentTransaction.get();
23 if (transaction == null || transaction.getState().equals(
24 BasicTransactionSemantics.State.CLOSED)) {
25 transaction = createTransaction();
26 currentTransaction.set(transaction);
27 }
28 return transaction;
29 }
上面我们可以看出来,如果transaction还未初始化或者transaction的状态是CLOSED(就是执行了close()方法改了状态),说明需要通过createTransaction()新建一个Transaction,createTransaction()这个方法在子类中实现的。我们来看看mem和file的createTransaction()方法的代码,先看mem的:
1 @Override
2 protected BasicTransactionSemantics createTransaction() {
3 if(!open) {
4 String msg = "Channel closed " + channelNameDescriptor;
5 if(startupError != null) {
6 msg += ". Due to " + startupError.getClass().getName() + ": " +
7 startupError.getMessage();
8 throw new IllegalStateException(msg, startupError);
9 }
10 throw new IllegalStateException(msg);
11 }
12 FileBackedTransaction trans = transactions.get();
13 if(trans != null && !trans.isClosed()) { //在这保证put和take只能一个时刻有一个
14 Preconditions.checkState(false,
15 "Thread has transaction which is still open: " +
16 trans.getStateAsString() + channelNameDescriptor);
17 }
18 trans = new FileBackedTransaction(log, TransactionIDOracle.next(),
19 transactionCapacity, keepAlive, queueRemaining, getName(),
20 channelCounter);
21 transactions.set(trans);
22 return trans;
23 }
这个就比mem的复杂了点,毕竟代码多了不少。
在看上面的getTransaction()方法,如果已经创建了一个Transaction则会放入currentTransaction中,然后以后再调用getTransaction()就会通过currentTransaction返回currentTransaction.get(),这莫不是同一个Transaction吗?那就好像有点不对了,对吧,那到底是怎么回事呢?
关键在于currentTransaction这个东西,我们看声明:private ThreadLocal<BasicTransactionSemantics> currentTransaction = new ThreadLocal<BasicTransactionSemantics>()是ThreadLocal的实例,可能有很多人不了解这个东西,其实我也不了解!!简单来说:ThreadLocal使得各线程能够保持各自独立的一个对象,ThreadLocal并不是一个Thread,而是Thread的局部变量,为解决多线程程序的并发问题提供了一种新的思路,详细请谷歌、百度之。ThreadLocal有一个ThreadLocalMap静态内部类,你可以简单理解为一个MAP,这个‘Map’为每个线程复制一个变量的‘拷贝’存储其中,这个“Map”的key是当前线程的ID,value就是set的变量,而get方法会依据当前线程ID从ThreadLocalMap中获取对应的变量,咱们这里就是Transaction。这下明白了吧,每个source和sink都会有单独的线程来驱动的,所以都有各自的Transaction,是不同的,因此也就可以并发了(针对memory channel)。
但是上面file的createTransaction()方法为什么是那样的?因为我们说了file的Transaction不能同时put和take(同一个Transaction一般只会做一个事就是put或者take),也就是不存在并发性的,所以在file channel中的transactions也设置为了private final ThreadLocal<FileBackedTransaction> transactions =new ThreadLocal<FileBackedTransaction>(),由于channel也是单独的线程驱使,所以这个transactions中始终只存在一对值就是file channel的线程ID和创建的Transaction,如果不同sink或者source调用getTransaction()时会试图通过createTransaction()方法来创建新的Transaction但是file的createTransaction()方法由于已经有了一个Transaction,在其关闭之前是不会同意 再次创建的,所以就只能等待这个Transaction关闭了,因此也就保证了put和take不会同时存在了。也就没有并发性了,性能自然大受影响。
那file channel为什么不让put和take同时操作呢?这个问题很值得研究,一:put、take、commit、rollback都会获取log的共享锁,一旦获取其他就只能读,获取锁的目的就是这四个操作都要写入log文件;二,put操作并不会将写入log的event的指针放到queue中,而是在commit中才会放到queue中;三、take的操作会直接从queue中取数据,这时如果put已经commit就可以获取数据,如果没有则会返回null;四、由于四个操作都会获取log锁,导致实际上写达不到并发,而且这个log锁使得即使是写不同的数据文件也不可能,因为只有这一个锁,不是每个数据文件一个锁(数据文件的个数是动态的这个不好做);五、若take和put同时操作会使得可能交替执行获取锁,此时可能put没commit而queue中无数据,take获取锁之后也没什么意义而且也是轮流不是并行,只会降低put和take的性能,比如put和take各自单独只需1s即可,但是这样可能需要2s甚至更长时间(take一直在等待put的commit)才能完成。综上不让put和take同时操作比较合理。
但是有没有更好的方案可以提高file的性能呢?因为file是基于文件的性能不可能很高,更为合理的办法是合理提高并发性,可以优化的一个方案是put、take、commit、rollback单独以文件存放,并设置相应的多个锁,但是文件的动态变化以及put和put、take和take、commit和commit、rollback和rollback之间的并发性又难以实现了,似乎只适合take和put的并发,这样貌似会使得file channel更复杂了,但是性能应该会提高一些,会不会得不偿失啊?
还有一个问题就是:file channel中的createTransaction()方法如果再次创建Transaction,而先前创建的并未关闭,会执行Preconditions.checkState(false,"Thread has transaction which is still open: " +trans.getStateAsString()+ channelNameDescriptor)会直接抛出异常,但是似乎日志中没有类似的异常啊,而且进程也并未中断,但是显然使用了file channel的flume,sink和source可以正常运行,这是怎么搞得?