Java中有哪些隊列
- ArrayBlockingQueue 使用ReentrantLock
- LinkedBlockingQueue 使用ReentrantLock
- ConcurrentLinkedQueue 使用CAS
- 等等
我們清楚使用鎖的性能比較低，盡量使用無鎖設計。接下來就我們來認識下Disruptor。
Disruptor簡單使用
github地址：https://github.com/LMAX-Exchange/disruptor/wiki/Performance-Results
先簡單介紹下：
- Disruptor它是一個開源的并發框架，并獲得2011 Duke’s程序框架創新獎【Oracle】，能夠在無鎖的情況下實現網絡的Queue并發操作。英國外匯交易公司LMAX開發的一個高性能隊列，號稱單線程能支撐每秒600萬訂單~
- 日志框架Log4j2 異步模式采用了Disruptor來處理
- 局限呢，他就是個內存隊列，也就是說無法支撐分布式場景。
簡單使用
數據傳輸對象
@Data
public class EventData {
private Long value;
}
消費者
public class EventConsumer implements WorkHandler<EventData> {
/**
* 消費回調
* @param eventData
* @throws Exception
*/
@Override
public void onEvent(EventData eventData) throws Exception {
Thread.sleep(5000);
System.out.println(Thread.currentThread() + ", eventData:" + eventData.getValue());
}
}生產者
public class EventProducer {
private final RingBuffer<EventData> ringBuffer;
public EventProducer(RingBuffer<EventData> ringBuffer) {
this.ringBuffer = ringBuffer;
}
public void sendData(Long v){
// cas展位
long next = ringBuffer.next();
try {
EventData eventData = ringBuffer.get(next);
eventData.setValue(v);
} finally {
// 通知等待的消費者
System.out.println("EventProducer send success, sequence:"+next);
ringBuffer.publish(next);
}
}
}測試類
public class DisruptorTest {
public static void main(String[] args) {
// 2的n次方
int bufferSize = 8;
Disruptor<EventData> disruptor = new Disruptor<EventData>(
() -> new EventData(), // 事件工廠
bufferSize, // 環形數組大小
Executors.defaultThreadFactory(), // 線程池工廠
ProducerType.MULTI, // 支持多事件發布者
new BlockingWaitStrategy()); // 等待策略
// 設置消費者
disruptor.handleEventsWithWorkerPool(
new EventConsumer(),
new EventConsumer(),
new EventConsumer(),
new EventConsumer());
disruptor.start();
RingBuffer<EventData> ringBuffer = disruptor.getRingBuffer();
EventProducer eventProducer = new EventProducer(ringBuffer);
long i = 0;
for(;;){
i++;
eventProducer.sendData(i);
try {
Thread.sleep(1500);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}核心組件
基于上面簡單例子來看確實很簡單，Disruptor幫我們封裝好了生產消費模型的實現，接下來我們來看下他是基于哪些核心組件來支撐起一個高性能無鎖隊列呢？
RingBuffer： 環形數組，底層使用數組entries，在初始化時填充數組，避免不斷新建對象帶來的開銷。后續只會對entries做更新操作
Sequencer： 核心管家
- 定義生產同步的實現：SingleProducerSequencer?單生產、MultiProducerSequencer多生產
- 當前寫的進度Sequence cursor
- 所有消費者進度的數組Sequence[] gatingSequences
- MultiProducerSequencer?可用區availableBuffer【利用空間換取查詢效率】
Sequence： 本身就是一個序號器用來標識處理進度，也可以當做是一個atomicInteger; 還有另外一個特點，為了解決偽共享問題而引入的：緩存行填充。這個在后面介紹。
workProcessor: 處理Event的循環，在循環中獲取Disruptor的事件，然后把事件分配給各個handler
EventHandler: 負責業務邏輯的handler,自己實現。
WaitStrategy： 消費者 如何等待 事件的策略，定義了如下策略
- leepingWaitStrategy：自旋 + yield + sleep
- BlockingWaitStrategy：加鎖，適合CPU資源緊張（不需要切換線程），系統吞吐量無要求的
- YieldingWaitStrategy：自旋 + yield + 自旋
- BusySpinWaitStrategy：自旋，減少線程之前切換
- PhasedBackoffWaitStrategy：自旋 + yield + 自定義策略
帶著問題來解析代碼？
1、多生產者如何保證消息生產不會相互覆蓋?！救绾芜_到互斥效果】
每個線程獲取不同的一段數組空間，然后通過CAS判斷這段空間是否已經分配出去。
接下來我們看下多生產類MultiProducerSequencer中next方法【獲取生產序號】
// 消費者上一次消費的最小序號 // 后續第二點會講到
private final Sequence gatingSequenceCache = new Sequence(Sequencer.INITIAL_CURSOR_VALUE);
// 當前進度的序號
protected final Sequence cursor = new Sequence(Sequencer.INITIAL_CURSOR_VALUE);
// 所有消費者的序號 //后續第二點會講到
protected volatile Sequence[] gatingSequences = new Sequence[0];
public long next(int n)
{
if (n < 1)
{
throw new IllegalArgumentException("n must be > 0");
}
long current;
long next;
do
{
// 當前進度的序號，Sequence的value具有可見性，保證多線程間線程之間能感知到可申請的最新值
current = cursor.get();
// 要申請的序號空間:最大序列號
next = current + n;
long wrapPoint = next - bufferSize;
// 消費者最小序列號
long cachedGatingSequence = gatingSequenceCache.get();
// 大于一圈 || 最小消費序列號>當前進度
if (wrapPoint > cachedGatingSequence || cachedGatingSequence > current)
{
long gatingSequence = Util.getMinimumSequence(gatingSequences, current);
// 說明大于1圈，并沒有多余空間可以申請
if (wrapPoint > gatingSequence)
{
LockSupport.parkNanos(1); // TODO, should we spin based on the wait strategy?
continue;
}
// 更新最小值到Sequence的value中
gatingSequenceCache.set(gatingSequence);
}
// CAS成功后更新當前Sequence的value
else if (cursor.compareAndSet(current, next))
{
break;
}
}
while (true);
return next;
}
2、生產者向序號器申請寫的序號，如序號正在被消費，Sequencer是如何知道哪些序號是可以被寫入的呢？【未消費則被覆蓋如何處理】
從gatingSequences中取得最小的序號，生產者最多能寫到這個序號的后一位。通俗來講就是申請的序號不能大于最小消費者序號一圈【申請到最大序列號-buffersize 要小于/等于 最小消費的序列號】的時候， 才能申請到當前寫的序號
public final EventHandlerGroup<T> handleEventsWithWorkerPool(final WorkHandler<T>... workHandlers)
{
return createWorkerPool(new Sequence[0], workHandlers);
}
EventHandlerGroup<T> createWorkerPool(
final Sequence[] barrierSequences, final WorkHandler<? super T>[] workHandlers)
{
final SequenceBarrier sequenceBarrier = ringBuffer.newBarrier(barrierSequences);
final WorkerPool<T> workerPool = new WorkerPool<>(ringBuffer, sequenceBarrier, exceptionHandler, workHandlers);
consumerRepository.add(workerPool, sequenceBarrier);
final Sequence[] workerSequences = workerPool.getWorkerSequences();
updateGatingSequencesForNextInChain(barrierSequences, workerSequences);
return new EventHandlerGroup<>(this, consumerRepository, workerSequences);
}
private void updateGatingSequencesForNextInChain(final Sequence[] barrierSequences, final Sequence[] processorSequences)
{
if (processorSequences.length > 0)
{
// 消費者啟動后就會將所有消費者存放入AbstractSequencer中gatingSequences
ringBuffer.addGatingSequences(processorSequences);
for (final Sequence barrierSequence : barrierSequences)
{
ringBuffer.removeGatingSequence(barrierSequence);
}
consumerRepository.unMarkEventProcessorsAsEndOfChain(barrierSequences);
}
}
3、在多生產者情況下，生產者是申請到一段可寫入的序號，然后再寫入這些序號中，那么消費者是如何感知哪些序號是可以被消費的呢？【借問提1圖說明】
這個前提是多生產者情況下，第一點我們說過每個線程獲取不同的一段數組空間，那么現在單單通過序號已經不夠用了，MultiProducerSequencer?使用了int 數組 【availableBuffer?】來標識當前序號是否可用。當生產者成功生產事件后會將availableBuffer中當前序列號置為1標識可以讀取。
如此消費者可以讀取的的最大序號就是我們availableBuffer中第一個不可用序號-1。
初始化availableBuffer流程
public MultiProducerSequencer(int bufferSize, final WaitStrategy waitStrategy)
{
super(bufferSize, waitStrategy);
// 初始化可用數組
availableBuffer = new int[bufferSize];
indexMask = bufferSize - 1;
indexShift = Util.log2(bufferSize);
initialiseAvailableBuffer();
}
// 初始化默認availableBuffer為-1
private void initialiseAvailableBuffer()
{
for (int i = availableBuffer.length - 1; i != 0; i--)
{
setAvailableBufferValue(i, -1);
}
setAvailableBufferValue(0, -1);
}
// 生產者成功生產事件將可用區數組置為1
public void publish(final long sequence)
{
setAvailable(sequence);
waitStrategy.signalAllWhenBlocking();
}
private void setAvailableBufferValue(int index, int flag)
{
long bufferAddress = (index * SCALE) + BASE;
UNSAFE.putOrderedInt(availableBuffer, bufferAddress, flag);
}
消費者消費流程
WorkProcessor類中消費run方法
public void run()
{
boolean processedSequence = true;
long cachedAvailableSequence = Long.MIN_VALUE;
long nextSequence = sequence.get();
T event = null;
while (true)
{
try
{
// 先通過cas獲取消費事件的占有權
if (processedSequence)
{
processedSequence = false;
do
{
nextSequence = workSequence.get() + 1L;
sequence.set(nextSequence - 1L);
}
while (!workSequence.compareAndSet(nextSequence - 1L, nextSequence));
}
// 數據就緒，可以消費
if (cachedAvailableSequence >= nextSequence)
{
event = ringBuffer.get(nextSequence);
// 觸發回調函數
workHandler.onEvent(event);
processedSequence = true;
}
else
{
// 獲取可以被讀取的下標
cachedAvailableSequence = sequenceBarrier.waitFor(nextSequence);
}
}
// ....省略
}
notifyShutdown();
running.set(false);
}
public long waitFor(final long sequence)
throws AlertException, InterruptedException, TimeoutException
{
checkAlert();
// 這個值獲取的current write 下標，可以認為全局消費下標。此處與每一段的write1和write2下標區分開
long availableSequence = waitStrategy.waitFor(sequence, cursorSequence, dependentSequence, this);
if (availableSequence < sequence)
{
return availableSequence;
}
// 通過availableBuffer篩選出第一個不可用序號 -1
return sequencer.getHighestPublishedSequence(sequence, availableSequence);
}
public long getHighestPublishedSequence(long lowerBound, long availableSequence)
{
// 從current read下標開始， 循環至 current write,如果碰到availableBuffer 為-1 直接返回
for (long sequence = lowerBound; sequence <= availableSequence; sequence++)
{
if (!isAvailable(sequence))
{
return sequence - 1;
}
}
return availableSequence;
}
解決偽共享問題
什么是偽共享問題呢？
為了提高CPU的速度，Cpu有高速緩存Cache,該緩存最小單位為緩存行CacheLine,他是從主內存復制的Cache的最小單位，通常是64字節。一個Java的long類型是8字節，因此在一個緩存行中可以存8個long類型的變量。如果你訪問一個long數組，當數組中的一個值被加載到緩存中，它會額外加載另外7個。因此你能非?？斓乇闅v這個數組。
偽共享問題是指，當多個線程共享某份數據時，線程1可能拉到線程2的數據在其cache line中，此時線程1修改數據，線程2取其數據時就要重新從內存中拉取，兩個線程互相影響，導致數據雖然在cache line中，每次卻要去內存中拉取。
Disruptor是如何解決的呢？
在value前后統一都加入7個Long類型進行填充，線程拉取時，不論如何都會占滿整個緩存
回顧總結：Disuptor為何能稱之為高性能的無鎖隊列框架呢？
- 緩存行填充，避免緩存頻繁失效?！緅ava8中也引入@sun.misc.Contended注解來避免偽共享】
- 無鎖競爭：通過CAS 【二階段提交】
- 環形數組：數據都是覆蓋，避免GC
- 底層更多的使用位運算來提升效率
