LinkedBlockingQueue 是很优秀的阻塞队列实现，某些资料显示可以达到50w QPS左右，其读写使用两把锁，但底层使用ReentrantLock 进行阻塞，虽然也是基于CAS原理，但多线程操作时不可避免也有等待时间，所以在考虑想如何进一步提升其吞吐量和QPS，联想了分段锁和kafka分区后，觉得可以增加队列数量降低锁等待的概率，例如10个线程同时获取一个队列的写锁，其中9个线程在等待。但分成10个队列去存储数据，那么每把锁可能就没有锁竞争或者只有两、三个线程参与同一把锁竞争，可以大大降低锁等待时间。
普通入队实现：
import java.util.concurrent.LinkedBlockingQueue;
public class BlockQueue {
private LinkedBlockingQueue<String> queue = null ;
public void init(int size){
queue = new LinkedBlockingQueue( size ) ;
}
public void offer (String data){
queue.offer(data) ;
}
}
分区入队实现：
import java.util.concurrent.LinkedBlockingQueue;
public class PartitionBlockQueue {
private LinkedBlockingQueue<String>[] queue = null ;
public void init(int size , int partition ){
queue = new LinkedBlockingQueue[partition] ;
for(int i = 0 ;i< partition ;i++){
queue[i] = new LinkedBlockingQueue<>(size/partition);
}
}
public void offer (String data){
int idx = Math.abs(data.hashCode())%queue.length;
queue[idx].offer(data);
}
}
测试代码,多线程执行1000w个元素入队
public static void testNormalBlock(int queueSize , int thread) throws InterruptedException {
long start = System.currentTimeMillis() ;
BlockQueue b = new BlockQueue();
b.init(queueSize);
CountDownLatch countDownLatch = new CountDownLatch(thread);
for(int i=0;i<thread;i++){
Thread t = new Thread(()->{
System.out.println(Thread.currentThread().getId()+"Thread run" );
for(int ix = 0 ;ix< queueSize/thread ;ix++){
b.offer(String.valueOf(ix));
}
System.out.println(Thread.currentThread().getId()+"Thread end" );
countDownLatch.countDown();
});
t.start();
}
countDownLatch.await();
long end = System.currentTimeMillis() ;
System.out.println((end - start)+" ms" );
}
public static void testPartitionBlock(int queueSize , int partition ,int thread) throws InterruptedException {
long start = System.currentTimeMillis() ;
PartitionBlockQueue pb = new PartitionBlockQueue();
pb.init(queueSize,partition);
CountDownLatch countDownLatch = new CountDownLatch(thread);
for(int i=0;i<thread;i++){
Thread t = new Thread(()->{
System.out.println(Thread.currentThread().getId()+"Thread run" );
for(int ix = 0 ;ix< queueSize/thread ;ix++){
pb.offer(String.valueOf(ix));
}
System.out.println(Thread.currentThread().getId()+"Thread end" );
countDownLatch.countDown();
});
t.start();
}
countDownLatch.await();
long end = System.currentTimeMillis() ;
System.out.println((end - start)+" ms" );
}
运行结果
testNormalBlock
/*
threads 10 -> 4121 ms 5472 ms 5229 ms 5218 ms
threads 20 -> 3841 ms 5058 ms 4967 ms 5155 ms
threads 50 -> 5359 ms 5590 ms 5308 ms
threads 100 -> 5168 ms 5188 ms 5551 ms
*/
testPartitionBlock
/*
Threads 10 partition 10 -> 2171 ms 2210 ms 2114 ms 2150 ms
Threads 20 partition 10 ->2051 ms 1914 ms 1957 ms 1966 ms
threads 50 parttion 10 -> 2290 ms 2326 ms 2373 ms 2435 ms
threads 100 parttion 10 -> 1854 ms 2415 ms 2636 ms 2417 ms
Threads 10 partition 20 ->868 ms 839 ms 905 ms
Threads 20 partition 20 ->904 ms 1745 ms 1745 ms
*/
结论：使用分区后性能提升很大，吞吐量数倍提升。且随着分区数增大
