本文共 6368 字，大约阅读时间需要 21 分钟。
使用LinkedBlockingDeque实现,
数据基于双向队列存储，可从从两端取出，插入。继承BasicTransactionSemantics,实现doPut，doTake，doCommit，doRollback方法。
源码:private class MemoryTransaction extends BasicTransactionSemantics { //和MemoryChannel一样，内部使用LinkedBlockingDeque来保存没有commit的Event private LinkedBlockingDeque takeList; private LinkedBlockingDeque putList; private final ChannelCounter channelCounter; //下面两个变量用来表示put的Event的大小、take的Event的大小 private int putByteCounter = 0; private int takeByteCounter = 0; public MemoryTransaction(int transCapacity, ChannelCounter counter) { //用transCapacity来初始化put、take的队列 putList = new LinkedBlockingDeque (transCapacity); takeList = new LinkedBlockingDeque (transCapacity); channelCounter = counter; } @Override protected void doPut(Event event) throws InterruptedException { //doPut操作，先判断putList中是否还有剩余空间，有则把Event插入到该队列中，同时更新putByteCounter //没有剩余空间的话，直接报ChannelException channelCounter.incrementEventPutAttemptCount(); int eventByteSize = (int)Math.ceil(estimateEventSize(event)/byteCapacitySlotSize); if (!putList.offer(event)) { throw new ChannelException( "Put queue for MemoryTransaction of capacity " + putList.size() + " full, consider committing more frequently, " + "increasing capacity or increasing thread count"); } putByteCounter += eventByteSize; } @Override protected Event doTake() throws InterruptedException { //doTake操作，首先判断takeList中是否还有剩余空间 channelCounter.incrementEventTakeAttemptCount(); if(takeList.remainingCapacity() == 0) { throw new ChannelException("Take list for MemoryTransaction, capacity " + takeList.size() + " full, consider committing more frequently, " + "increasing capacity, or increasing thread count"); } //然后判断，该MemoryChannel中的queue中是否还有空间，这里通过信号量来判断 if(!queueStored.tryAcquire(keepAlive, TimeUnit.SECONDS)) { return null; } Event event; //从MemoryChannel中的queue中取出一个event synchronized(queueLock) { event = queue.poll(); } Preconditions.checkNotNull(event, "Queue.poll returned NULL despite semaphore " + "signalling existence of entry"); //放到takeList中，然后更新takeByteCounter变量 takeList.put(event); int eventByteSize = (int)Math.ceil(estimateEventSize(event)/byteCapacitySlotSize); takeByteCounter += eventByteSize; return event; } @Override protected void doCommit() throws InterruptedException { //该对应一个事务的提交 //首先判断putList与takeList的相对大小 int remainingChange = takeList.size() - putList.size(); //如果takeList小，说明向该MemoryChannel放的数据比取的数据要多，所以需要判断该MemoryChannel是否有空间来放 if(remainingChange < 0) { // 1. 首先通过信号量来判断是否还有剩余空间 if(!bytesRemaining.tryAcquire(putByteCounter, keepAlive, TimeUnit.SECONDS)) { throw new ChannelException("Cannot commit transaction. Byte capacity " + "allocated to store event body " + byteCapacity * byteCapacitySlotSize + "reached. Please increase heap space/byte capacity allocated to " + "the channel as the sinks may not be keeping up with the sources"); } // 2. 然后判断，在给定的keepAlive时间内，能否获取到充足的queue空间 if(!queueRemaining.tryAcquire(-remainingChange, keepAlive, TimeUnit.SECONDS)) { bytesRemaining.release(putByteCounter); throw new ChannelFullException("Space for commit to queue couldn't be acquired." + " Sinks are likely not keeping up with sources, or the buffer size is too tight"); } } int puts = putList.size(); int takes = takeList.size(); //如果上面的两个判断都过了，那么把putList中的Event放到该MemoryChannel中的queue中。 synchronized(queueLock) { if(puts > 0 ) { while(!putList.isEmpty()) { if(!queue.offer(putList.removeFirst())) { throw new RuntimeException("Queue add failed, this shouldn't be able to happen"); } } } //清空本次事务中用到的putList与takeList，释放资源 putList.clear(); takeList.clear(); } //更新控制queue大小的信号量bytesRemaining，因为把takeList清空了，所以直接把takeByteCounter加到bytesRemaining中。 bytesRemaining.release(takeByteCounter); takeByteCounter = 0; putByteCounter = 0; //因为把putList中的Event放到了MemoryChannel中的queue，所以把puts加到queueStored中去。 queueStored.release(puts); //如果takeList比putList大，说明该MemoryChannel中queue的数量应该是减少了，所以把(takeList-putList)的差值加到信号量queueRemaining if(remainingChange > 0) { queueRemaining.release(remainingChange); } if (puts > 0) { channelCounter.addToEventPutSuccessCount(puts); } if (takes > 0) { channelCounter.addToEventTakeSuccessCount(takes); } channelCounter.setChannelSize(queue.size()); } @Override protected void doRollback() { //当一个事务失败时，会进行回滚，即调用本方法 //首先把takeList中的Event放回到MemoryChannel中的queue中。 int takes = takeList.size(); synchronized(queueLock) { Preconditions.checkState(queue.remainingCapacity() >= takeList.size(), "Not enough space in memory channel " + "queue to rollback takes. This should never happen, please report"); while(!takeList.isEmpty()) { queue.addFirst(takeList.removeLast()); } //然后清空putList putList.clear(); } //因为清空了putList，所以需要把putList所占用的空间大小添加到bytesRemaining中 bytesRemaining.release(putByteCounter); putByteCounter = 0; takeByteCounter = 0; //因为把takeList中的Event回退到queue中去了，所以需要把takeList的大小添加到queueStored中 queueStored.release(takes); channelCounter.setChannelSize(queue.size()); } } 转载地址：http://qfkai.baihongyu.com/