深入java多线程与高并发：JMH与Disruptor，确定能学会？

<?xml version="1.0" encoding="UTF-8"?><project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><encoding>UTF-8</encoding><java.version>1.8</java.version><maven.compiler.source>1.8</maven.compiler.source><maven.compiler.target>1.8</maven.compiler.target></properties><groupId>mashibing.com</groupId><artifactId>HelloJMH2</artifactId><version>1.0-SNAPSHOT</version><dependencies><!-- https://mvnrepository.com/artifact/org.openjdk.jmh/jmh-core --><dependency><groupId>org.openjdk.jmh</groupId><artifactId>jmh-core</artifactId><version>1.21</version></dependency><!-- https://mvnrepository.com/artifact/org.openjdk.jmh/jmh-
generator-annprocess --><dependency><groupId>org.openjdk.jmh</groupId><artifactId>jmh-generator-annprocess</artifactId><version>1.21</version><scope>test</scope></dependency></dependencies></project>

package com.mashibing.jmh;import java.util.ArrayList;import java.util.List;import java.util.Random;public class PS {static List<Integer> nums = new ArrayList<>();static {
Random r = new Random();for (int i = ; i < ; i++) nums.add( +
r.nextInt());
}static void foreach() {
nums.forEach(v->isPrime(v));
}static void parallel() {
nums.parallelStream().forEach(PS::isPrime);
}static boolean isPrime(int num) {for(int i=; i<=num/; i++) {if(num % i == ) return false;
}return true;
}
}

package com.mashibing.jmh;import org.openjdk.jmh.annotations.Benchmark;import static org.junit.jupiter.api.Assertions.*;public class PSTest {@Benchmark@Warmup（iteration=, time=）//在专业测试里面首先要进行预热，预热多少次，预热多少时间@Fork()//意思是用多少个线程去执行我们的程序@BenchmarkMode(Mode.Throughput)//是对基准测试的一个模式，这个模式用的最多的是Throughput吞吐量@Measurement(iteration=, time=)//是整个测试要测试多少遍，调用这个方法要调用多少次
public void testForEach() {PS.foreach();
}
}

package com.mashibing.disruptor;public class LongEvent{private long value;public void set(long value){this.value = value;
}
@Overridepublic String toString(){return "LongEvent{" +"value=" + value +"}";
}
}

package com.mashibing.disruptor;import com.lmax.disruptor.EventFactory;public class LongEventFactory implements EventFactiry<LongEvent>{@Overridepublic LongEvent newInstance(){return new LongEvent();
}
}

package com.mashibing.disruptor;import com.lmax.disruptor.EventHandler;public class LongEventHandler implements EventHandler<LongEvent>{/**
*
*@param event
*@param sequence RingBuffer的序号
*@param endOfBatch 是否为最后一个元素
*@throws Exception
**/public static long count = ;
@Overridepublic void onEvent(LongEvent event,long sequence,boolean endOfBatch) throwsException{
count++;
System.out.println("["+Thread.currentThread().getName()+"]"+event+"序
号："+sequence);
}
}

package com.mashibing.disruptor;import java.util.concurrent.Executor;import java.util.concurrent.Executors;import com.lmax.disruptor.dsl.Disruptor;import com.lmax.disruptor.RingBuffer;import com.lmax.disruptor.util.DaemonThreadFactory;import java.nio.ByteBuffer;public class Main01{public static void main(String[] args) thrwos Exception{//the factory for the eventLongEvenFactory factory = new LongEventFactory();//Specify the of the ring buffer,must be power of 2.int bufferSize = ;//Construct the DisruptorDisruptor<LongEvent> disruptor = new Disruptor<>(factory,bufferSize,
Executors.defaultThreadFactory());//Connect the handlerdisruptor.handleEventsWith(new LongEventHandler());//Start the Disruptor,start all threads runningdisruptor.start();//Get the ring buffer form the Disruptor to be used for publishing.RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();//官方例程long sequence = ringBuffer.next();//Grab the next sequencetry{
LongEvent event=ringBuffer.get(sequence);//Get the entry in theDisruptor//for the sequenceevent.set(8888L);//Fill with data}finally{
ringBuffer.publish(sequence);
}
}
}

package com.mashibing.disruptor;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.util.DaemonThreadFactory;
import java.nio.ByteBuffer;public class Main02{public static void main(String[] args) thrwos Exception{//the factory for the eventLongEvenFactory factory = new LongEventFactory();//Specify the of the ring buffer,must be power of 2.int bufferSize = ;//Construct the DisruptorDisruptor<LongEvent> disruptor = new Disruptor<>(factory,bufferSize,
DaemonThreadFactory.INSTANCE);//Connect the handlerdisruptor.handleEventsWith(new LongEventHandler());//Start the Disruptor,start all threads runningdisruptor.start();//Get the ring buffer form the Disruptor to be used for publishing.RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();//========================================================================EventTranslator<LongEvent> translator1 = new EventTranslator<LongEvent>
(){
@Overridepublic void translateTo(LongEvent event,long sequence){event.set(8888L); }
};
ringBuffer.publishEvent(translator1);//========================================================================EventTranslatorOneArg<LongEvent,Long> translator2 = newEventTranslatorOneArg<LongEvent,Long>(){
@Overridepublic void translateTo(LongEvent event,long sequence,Long l){event.set(l); }
};
ringBuffer.publishEvent(translator2,7777L);//========================================================================EventTranslatorTwoArg<LongEvent,Long,Long> translator3 = newEventTranslatorTwoArg<LongEvent,Long,Long>(){
@Overridepublic void translateTo(LongEvent event,long sequence,Long l1,Long l2){ event.set(l); }
};
ringBuffer.publishEvent(translator3,10000L,10000L);//========================================================================EventTranslatorThreeArg<LongEvent,Long,Long,Long> translator4 = newEventTranslatorThreeArg<LongEvent,Long,Long,Long>(){
@Overridepublic void translateTo(LongEvent event,long sequence,Long l1,Long
l2,Long l3){ event.set(l1+ l2+ l3); }
};
ringBuffer.publishEvent(translator4,10000L,10000L,10000L);//========================================================================EventTranslatorVararg<LongEvent> translator5 = newEventTranslatorThreeArg<LongEvent>(){
@Overridepublic void translateTo(LongEvent event,long sequence,Object...
objects){long result = ;for(Object o : objects){long l =(Long)o;
result +=l;
}
}
};
ringBuffer.publishEvent(translator5,10000L,10000L,10000L,10000L);
}
}

package com.mashibing.disruptor;import java.util.concurrent.Executor;import java.util.concurrent.Executors;import com.lmax.disruptor.dsl.Disruptor;import com.lmax.disruptor.RingBuffer;import com.lmax.disruptor.util.DaemonThreadFactory;import java.nio.ByteBuffer;public class Main03{public static void main(String[] args) thrwos Exception{//the factory for the eventLongEvenFactory factory = new LongEventFactory();//Specify the of the ring buffer,must be power of 2.int bufferSize = 1024;//Construct the DisruptorDisruptor<LongEvent> disruptor = new Disruptor<>(LongEvent::new,
bufferSize,DaemonThreadFactory.INSTANCE);//Connect the handlerdisruptor.handleEventsWith((event,sequence,endOfBatch)-
>System.out.println("Event:"+event));//Start the Disruptor,start all threads runningdisruptor.start();//Get the ring buffer form the Disruptor to be used for publishing.RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
ringBuffer.publishEvent((event, sequence)-> event.set(10000L));System.in.read();
}
}

package com.mashibing.disruptor;import java.util.concurrent.Executor;import java.util.concurrent.Executors;import com.lmax.disruptor.dsl.Disruptor;import com.lmax.disruptor.RingBuffer;import com.lmax.disruptor.util.DaemonThreadFactory;import java.nio.ByteBuffer;public class Main04_ProducerType{public static void main(String[] args) thrwos Exception{//the factory for the eventLongEvenFactory factory = new LongEventFactory();//Specify the of the ring buffer,must be power of 2.int bufferSize = ;//Construct the Disruptor//Disruptor<LongEvent> disruptor = new Disruptor<>(factory,bufferSize,Executors.defaultThreadFactory());
Disruptor<LongEvent> disruptor = new Disruptor<>(factory,bufferSize,
Executors.defaultThreadFactory(),ProducerType.SINGLE,newBlockingWaitStrategy());//Connect the handlerdisruptor.handleEventsWith(new LongEventHandler());//Start the Disruptor,start all threads runningdisruptor.start();//Get the ring buffer form the Disruptor to be used for publishing.RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();//========================================================================final int threadCount = ;
CycliBarrier barrier=new CycliBarrier(threadCount);
ExecutorService service = Executors.newCachedThreadPool();for(long i=; i<threadCount; i++){final long threadNum = i;
service.submit(()->{
System.out.printf("Thread %s ready to start!\n",threadNum);try{
barrier.await();
}catch(InterruptedException e){
e.printStackTrace();
}catch(BrokenBarrierException e){
e.printStackTrace();
}for(int j=; j<;j++){
ringBuffer.publishEvent((event,sequence)->{
event.set(threadNum);
System.out.println("生产了"+threadNum);
});
}
});
}
service.shutdown();//disruptor.shutdown();TimeUnit.SECONDS.sleep();
System.out.println(LongEventHandler.count);
}
}

package com.mashibing.disruptor;import java.util.concurrent.Executor;import java.util.concurrent.Executors;import com.lmax.disruptor.dsl.Disruptor;import com.lmax.disruptor.RingBuffer;import com.lmax.disruptor.util.DaemonThreadFactory;import java.nio.ByteBuffer;public class Main05_WaitStrategy{public static void main(String[] args) thrwos Exception{//the factory for the eventLongEvenFactory factory = new LongEventFactory();//Specify the of the ring buffer,must be power of 2.int bufferSize = ;//Construct the DisruptorDisruptor<LongEvent> disruptor = new Disruptor<>(factory,bufferSize,
Executors.defaultThreadFactory(),ProducerType.MULTI,new SleepingWaitStrategy());//Connect the handlerdisruptor.handleEventsWith(new LongEventHandler());//Start the Disruptor,start all threads runningdisruptor.start();//Get the ring buffer form the Disruptor to be used for publishing.RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();//========================================================================final int threadCount = ;
CycliBarrier barrier=new CycliBarrier(threadCount);ExecutorService service = Executors.newCachedThreadPool();for(long i=; i<threadCount; i++){final long threadNum = i;
service.submit(()->{
System.out.printf("Thread %s ready to start!\n",threadNum);try{
barrier.await();
}catch(InterruptedException e){
e.printStackTrace();
}catch(BrokenBarrierException e){
e.printStackTrace();
}for(int j=; j<;j++){
ringBuffer.publishEvent((event,sequence)->{
event.set(threadNum);
System.out.println("生产了"+threadNum);
});
}
});
}
service.shutdown();//disruptor.shutdown();TimeUnit.SECONDS.sleep();
System.out.println(LongEventHandler.count);
}
}

package com.mashibing.disruptor;import java.util.concurrent.Executor;import java.util.concurrent.Executors;import com.lmax.disruptor.dsl.Disruptor;import com.lmax.disruptor.RingBuffer;import com.lmax.disruptor.util.DaemonThreadFactory;import java.nio.ByteBuffer;public class Main06_MultiConsumer{public static void main(String[] args) thrwos Exception{//the factory for the eventLongEvenFactory factory = new LongEventFactory();//Specify the of the ring buffer,must be power of 2.int bufferSize = ;//Construct the DisruptorDisruptor<LongEvent> disruptor = new Disruptor<>(factory,bufferSize,
Executors.defaultThreadFactory(),ProducerType.MULTI,new SleepingWaitStrategy());//Connect the handlersLongEventHandler h1 = new LongEventHandler();LongEventHandler h2 = new LongEventHandler();
disruptor.handleEventsWith(h1,h2);//Start the Disruptor,start all threads runningdisruptor.start();//Get the ring buffer form the Disruptor to be used for publishing.RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();//========================================================================final int threadCount = ;
CycliBarrier barrier=new CycliBarrier(threadCount);
ExecutorService service = Executors.newCachedThreadPool();for(long i=; i<threadCount; i++){final long threadNum = i;
service.submit(()->{
System.out.printf("Thread %s ready to start!\n",threadNum);try{
barrier.await();
}catch(InterruptedException e){
e.printStackTrace();
}catch(BrokenBarrierException e){
e.printStackTrace();
}for(int j=; j<;j++){
ringBuffer.publishEvent((event,sequence)->{
event.set(threadNum);
System.out.println("生产了"+threadNum);
});
}
});
}
service.shutdown();//disruptor.shutdown();TimeUnit.SECONDS.sleep();
System.out.println(LongEventHandler.count);
}
}

package com.mashibing.disruptor;import java.util.concurrent.Executor;import java.util.concurrent.Executors;import com.lmax.disruptor.dsl.Disruptor;import com.lmax.disruptor.RingBuffer;import com.lmax.disruptor.util.DaemonThreadFactory;import java.nio.ByteBuffer;public class Main07_ExceptionHandler{public static void main(String[] args) thrwos Exception{//the factory for the eventLongEvenFactory factory = new LongEventFactory();//Specify the of the ring buffer,must be power of 2.int bufferSize = ;//Construct the DisruptorDisruptor<LongEvent> disruptor = new Disruptor<>(factory,bufferSize,
Executors.defaultThreadFactory(),ProducerType.MULTI,new SleepingWaitStrategy());//Connect the handlersEventHandler h1 = (event,sequence,end)->{
System.out.println("消费者出异常")；
};
disruptor.handleEventsWith(h1);
disruptor.handleExceptionsFor(h1).with(newExceptionHandler<LongEvent>(){@Overridepublic void handleEventException(Throwable throwable,longl,LongEvent longEvent){
throwable.printStackTrace();
}@Overridepublic void handleOnStartException(Throwable throwable){
System.out.println("Exception Start to Handle!");
}@Overridepublic void handleOnShutdownException(Throwable throwable){
System.out.println("Exception End to Handle!");
}
});//Start the Disruptor,start all threads runningdisruptor.start();//Get the ring buffer form the Disruptor to be used for publishing.RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();//========================================================================final int threadCount = ;
CycliBarrier barrier=new CycliBarrier(threadCount);
ExecutorService service = Executors.newCachedThreadPool();for(long i=; i<threadCount; i++){final long threadNum = i;
service.submit(()->{
System.out.printf("Thread %s ready to start!\n",threadNum);try{
barrier.await();
}catch(InterruptedException e){
e.printStackTrace();
}catch(BrokenBarrierException e){
e.printStackTrace();
}for(int j=; j<;j++){
ringBuffer.publishEvent((event,sequence)->{
event.set(threadNum);
System.out.println("生产了"+threadNum);
});
}
});
}
service.shutdown();//disruptor.shutdown();TimeUnit.SECONDS.sleep();
System.out.println(LongEventHandler.count);
}
}