抽象并发控制类AbstractConcurrentControl ：
使用步骤：
1.单元测试类继承此类；
2.重写concurrentCode方法，定义自己的并发代码；
3.重写encapsulatingData方法，定义数据的拼装，并在concurrentCode方法中调用；
4.重写blockingMainThread方法，定义主线程阻塞策略
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
/**
* @author Yangzhen
* @Description 1.单元测试类继承此类；
* 2.重写concurrentCode方法，定义自己的并发代码；
* 3.重写encapsulatingData方法，并在concurrentCode方法中调用，定义数据的拼装；
* 4.重写blockingMainThread方法，定义主线程阻塞策略
* @date 2018-12-26 10:06
**/
public abstract class AbstractConcurrentControl {
public AtomicLong longCounter = new AtomicLong(0);
public AtomicInteger intCounter = new AtomicInteger(0);
/**
* 默认并发线程数为2000个，子类可重写
*/
private static final int DEFAULT_CONCURRENT_CONTROL = 2000;
private CountDownLatch blockLatch;
/**
* 并发线程数量，默认2000
*/
private int concurrentThreadNum;
private ThreadPoolExecutor threadPool;
public AbstractConcurrentControl() {
this(DEFAULT_CONCURRENT_CONTROL);
}
public AbstractConcurrentControl(int concurrentThreadNum) {
this.concurrentThreadNum = concurrentThreadNum;
blockLatch = new CountDownLatch(concurrentThreadNum);
threadPool = new ThreadPoolExecutor(concurrentThreadNum, concurrentThreadNum, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());
}
/**
* 并发执行线程
*
* @Title: process
* @date 2018年12月26日 上午11:19:20
* @author yz
*/
public final void process() {
for (int i = 0; i < concurrentThreadNum; i++) {
threadPool.submit(new Runnable() {
@Override
public void run() {
try {
blockLatch.await();
concurrentCode();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
// 最后一个线程时可以打个断点
if (i == concurrentThreadNum - 1) {
blockLatch.countDown();
} else {
blockLatch.countDown();
}
}
blockingMainThread();
}
/**
* 并发代码
*
* @Title: concurrentCode
* @date 2018年12月26日 下午2:05:25
* @author yz
*/
protected abstract void concurrentCode();
/**
* 并发数据
*
* @return
* @Title: encapsulatingData
* @date 2018年12月26日 下午2:06:14
* @author yz
*/
protected <T> T encapsulatingData() {
return null;
}
/**
* 阻塞主线程，防止JVM关闭，不建议使用Xxx.class.wait，可以使用TimeUnit.SECONDS.sleep(200);
* 如果使用TimeUnit.SECONDS.sleep(200)，要保证休眠时长足够跑完你的程序，否则会出现异常，因为JVM已经关闭，而测试的线程可能没有执行完成
*
* @Title: blockingMainThread
* @date 2018年12月26日 下午6:55:03
* @author yz
*/
protected void blockingMainThread() {
if (this.threadPool == null) {
return;
}
this.threadPool.shutdown();
while (!this.threadPool.isTerminated()) {
try {
TimeUnit.MILLISECONDS.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println("线程池关闭");
}
}
单元测试案例：
public class RequestTest extends AbstractConcurrentControl {
public RequestTest() {
super(1);
}
private MyService myService = new MyService();
/**
* 单元测试，调用process方法
*/
@Test
public void test() {
process();
}
@Override
public void concurrentCode() {
Data data = (Data) encapsulatingData();
myService.process(data);
}
/**
* 定义数据
*
* @param <T>
* @return
*/
@SuppressWarnings("unchecked")
@Override
public <T> T encapsulatingData() {
Data data = new Data();
String name = "a";
int i = intCounter.incrementAndGet();
data.setId(i);
data.setName(name + ":" + i);
return (T) data;
}
}
public class Data {
private Integer id;
private String name;
public Integer getId() {
return id;
}
public void setId(Integer id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
}
public class MyService {
public void process(Data data) {
System.out.println(data.getId() + "-" + data.getName());
}
}
本文内容不用于商业目的，如涉及知识产权问题，请权利人联系51Testing小编(021-64471599-8017)，我们将立即处理
