1.java并发包介绍
JDK5.0(JDK1.5更名后)以后的版本引入高级并发特性，大多数的特性在java.util.concurrent包中，是专门用于多线程编程的，充分利用了现代多处理器和多核心系统的功能以编写大规模并发应用程序。主要包括原子量、并发集合、同步器、可重入锁，并对线程池的构造提供了强力的支持
2.线程池
java.util.concurrent.Executors提供了一个 java.util.concurrent.Executor接口的实现用于创建线程池
多线程技术主要解决处理器单元内多个线程执行的问题，它可以显著减少处理器单元的闲置时间，增加处理器单元的吞吐能力。
假设服务器完成一项任务所需时间为：T1 创建线程时间，T2 在线程中执行任务的时间，T3 销毁线程时间。如果T1 + T3 远大于 T2，则可以采用线程池，以提高服务器性能,减少创建和销毁线程所需消耗的时间。
一个线程池由以下四个基本部分组成：
线程池技术正是关心如何缩短或调整T1,T3时间从而提高服务器程序性能的技术。它把T1，T3分别安排在服务器程序的启动和结束的时间段或者一些空闲的时间段，这样在服务器程序处理客户请求时，免去了线程创建和销毁的开销。
线程池不仅调整T1,T3产生的时间段，而且它还显著减少了创建线程的数目，看一个例子：
假设一个服务器一天要处理100000个请求，并且每个请求需要一个单独的线程完成。在线程池中，线程数一般是固定的，
一般线程池大小是远小于100000。所以利用线程池的服务器程序不会为了创建100000而在处理请求时浪费时间，从而提高效率。
线程池的五种创建方式
3.线程池的使用
以下用Fixed Thread Pool作为示范，提供一个使用参考
LogNumVo
package com.ithzk.threadpool;
/**
* 用作返回 执行的数量的
* @author hzk
* @date 2018/3/29
*/
public class LogNumVo {
private static final long serialVersiOnUID= -5541722936350755569L;
private Integer dataNum;
private Integer successNum;
private Integer waitNum;
public Integer getDataNum() {
return dataNum;
}
public void setDataNum(Integer dataNum) {
this.dataNum = dataNum;
}
public Integer getSuccessNum() {
return successNum;
}
public void setSuccessNum(Integer successNum) {
this.successNum = successNum;
}
public Integer getWaitNum() {
return waitNum;
}
public void setWaitNum(Integer waitNum) {
this.waitNum = waitNum;
}
}
DealObject
package com.ithzk.threadpool;
/**
* @author hzk
* @date 2018/3/29
*/
public class DealObject {
private Integer identifyId;
private String data;
public DealObject(Integer identifyId, String data) {
this.identifyId = identifyId;
this.data = data;
}
public DealObject() {
}
public Integer getIdentifyId() {
return identifyId;
}
public void setIdentifyId(Integer identifyId) {
this.identifyId = identifyId;
}
public String getData() {
return data;
}
public void setData(String data) {
this.data = data;
}
}
AbstractCalculateThread
package com.ithzk.threadpool; import java.util.Collection; import java.util.concurrent.Callable; import java.util.concurrent.CountDownLatch; /** * @author hzk * @date 2018/3/29 */ public class AbstractCalculateThreadimplements Callable { protected Collection insertList; protected CountDownLatch countd; protected String threadCode; protected String batchNumber; public Collection getInsertList() { return insertList; } public void setInsertList(Collection insertList) { this.insertList = insertList; } public CountDownLatch getCountd() { return countd; } public void setCountd(CountDownLatch countd) { this.countd = countd; } public String getThreadCode() { return threadCode; } public void setThreadCode(String threadCode) { this.threadCode = threadCode; } public String getBatchNumber() { return batchNumber; } public void setBatchNumber(String batchNumber) { this.batchNumber = batchNumber; } public AbstractCalculateThread() { super(); } public AbstractCalculateThread(Collection insertList, CountDownLatch countd, String threadCode,String batchNumber) { super(); this.insertList = insertList; this.countd = countd; this.threadCode = threadCode; this.batchNumber = batchNumber; } public String call() throws Exception { return null; } }
CalculateDealThread
package com.ithzk.threadpool; import java.util.Collection; import java.util.concurrent.CountDownLatch; /** * @author hzk * @date 2018/3/29 */ public class CalculateDealThread extends AbstractCalculateThread{ private ExecutorPool executorPool = SpringContextUtil.getBean(ExecutorPool.class); @Override public String call() throws Exception { try { System.out.println("========开始跑线程【"+threadCode+"】"); return executorPool.syncBatchDealObject(insertList,batchNumber); } catch (Exception e) { e.printStackTrace(); System.out.println("========开始跑线程【"+threadCode+"】:"+e.getMessage()); }finally { countd.countDown(); } return null; } public CalculateDealThread() { super(); } public CalculateDealThread(Collection insertList, CountDownLatch countd, String threadCode,String batchNumber) { super(insertList, countd, threadCode, batchNumber); } }
ExecutorPool
package com.ithzk.threadpool;
import java.util.*;
import java.util.concurrent.*;
/**
* @author hzk
* @date 2018/3/29
*/
public class ExecutorPool {
/**
* 模拟需要处理数据的大小
*/
private static final int ARRAY_COUNT = 50000;
/**
* 开启多线程处理的条件
*/
private static final int MULTI_THREAD_STARTCOUNT = 10000;
/**
* 批量处理的大小
*/
private static final int BATCH_DEAL_SIZE = 100;
/**
* 每次开启线程数量
*/
public static final int THREAD_POOL_NUM=10;
public static void main(String[] args){
testExecutorPool();
}
public static void testExecutorPool(){
ArrayList dealObjects = new ArrayList();
for (int i = 0;i MULTI_THREAD_STARTCOUNT) {
try {
System.out.println("===================dataNum > 1000 | Multiple Thread Run=======================");
// 每次新增处理的条数
int batchInsertSize = BATCH_DEAL_SIZE;
// 定义保存的线程池
ExecutorService executorInsert = Executors.newFixedThreadPool(THREAD_POOL_NUM);
// 定义保存过程中返回的线程执行返回参数
List> futureListIsert = new ArrayList>();
// 线程 修改list
List> listDealObjects = new ArrayList>();
List> listLiveSyncLogInsert = pointDateClassify(dealObjects, batchInsertSize, listDealObjects);
if (null != listLiveSyncLogInsert && !listDealObjects.isEmpty()) {
System.out.println("===================切割后的大小:"+listLiveSyncLogInsert.size()+"=======================");
//配合使用CountDownLatch为了保证在执行完所有子程序之后再执行主程序
CountDownLatch countd = new CountDownLatch(listLiveSyncLogInsert.size());
for (int j = 0; j insert = listLiveSyncLogInsert.get(j);
Future future = executorInsert.submit(new CalculateDealThread(insert.values(), countd,"executor_pool_test_thread", null));
futureListIsert.add(future);
}
}
// 等待线程执行完成
executorInsert.shutdown();
for (Future future : futureListIsert) {
String json = future.get();
if (null != json && !"".equals(json)) {
将返回的json格式数据转换为实体类 进行业务记录
LogNumVo logNumVo = JSON.toJavaObject(JSON.parseObject(json),LogNumVo.class);
successNum += logNumVo.getSuccessNum();
waitNum += logNumVo.getWaitNum();
}
}
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
}
/**
* 拆分线程数
* 假设集合中有50000个元素 则按照100个一组切分 可切分为500组
* 即每个线程一次处理一组(100个元素)
*
* @author
* @param lPostUploadIntegralList
* @param batchInsertSize
* @param listPostUploadIsert
*/
@SuppressWarnings("all")
public static List> pointDateClassify(List lPostUploadIntegralList,int batchInsertSize, List> listJSONObjectUpdate) {
List> listLiveSyncLogInsert = new Vector>();
// 新增数据list
List integralListInsert = lPostUploadIntegralList;
System.out.println("============integralListInsert.size()=====:" + integralListInsert.size());
// 拆分数据(拆成多个List)
int inserti = 0;
if (integralListInsert != null && integralListInsert.size() > 0) {
ConcurrentHashMap integralListIns = null;
for (int l = 0; l ();
}
integralListIns.put(integralListInsert.get(l).getIdentifyId(), integralListInsert.get(l));
inserti++;
if ((inserti % batchInsertSize) == 0) {
listLiveSyncLogInsert.add(integralListIns);
integralListIns = null;
} else {
// 最后100条或不足100条数据
if ((l + 1) == integralListInsert.size()) {
listLiveSyncLogInsert.add(integralListIns);
}
}
}
}
System.out.println("=============listPostUploadInsert.size()====:" + listLiveSyncLogInsert.size());
return listLiveSyncLogInsert;
}
/**
* 多线程保存数据至数据库
*/
public String syncBatchDealObject(Collection insertList,String batchNumber) {
int successNum = 0, waitNum = 0;
Date currentDate = new Date(System.currentTimeMillis());
for (DealObject dealObject : insertList) {
try {
int icount = syncDealObject(dealObject,currentDate);
if(icount > 0){
successNum ++;
}else {
waitNum ++;
}
} catch (Exception e) {
e.printStackTrace();
++waitNum;
}
}
LogNumVo logNum = new LogNumVo();
logNum.setDataNum(0);
logNum.setSuccessNum(successNum);
logNum.setWaitNum(waitNum);
// 将记录实体类转为json格式反馈给线程池
return JSON.toJSONString(logNum);
}
/**
* 处理数据业务
* @param dealObject
* @param currentDate
* @return
*/
private int syncDealObject(DealObject dealObject,Date currentDate){
int successNum = 0;
//业务处理逻辑
if(null != dealObject.getData()){
successNum++;
}
return successNum;
}
}
4.BlockingQueue
BlockingQueue也是java.util.concurrent下的主要用来控制线程同步的工具。主要的方法是：put、take一对阻塞存取；add、poll一对非阻塞存取。
插入：
add(anObject)
把anObject加到BlockingQueue里,如果BlockingQueue可以容纳,则返回true,否则抛出异常
offer(anObject)
把anObject加到BlockingQueue里,如果BlockingQueue可以容纳,则返回true,否则返回false.
put(anObject)
把anObject加到BlockingQueue里,如果BlockQueue没有空间,则调用此方法的线程被阻塞直到BlockingQueue里面有空间再继续.
读取：
poll(time)
取走BlockingQueue里排在首位的对象,若不能立即取出,则可以等time参数规定的时间,取不到时返回null
take()
取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到Blocking有新的对象被加入为止
其他：
int remainingCapacity()
返回理想情况下（没有内存和资源约束）此队列可接受并且不会被阻塞的附加元素数量。
该数量总是等于此队列的初始容量，小于队列的当前 size(返回队列剩余的容量)。
注意，不能总是通过检查 remainingcapacity 来断定试图插入一个元素是否成功，因为可能是另一个线程将插入或移除某个元
素。
boolean remove(Object o)
从队列移除元素，如果存在，即移除一个或者更多，队列改变了返回true
public boolean contains(Object o)
查看队列是否存在这个元素，存在返回true
int drainTo(Collection<&＃63; super E> c)
传入的集合中的元素，如果在队列中存在，那么将队列中的元素移动到集合中
int drainTo(Collection<&＃63; super E> c, int maxElements)
和上面方法的区别在于，制定了移动的数量
以下是一个BlockQueue的基本使用参考:
Producer
package com.ithzk.BlockingQueueTest;
import java.util.concurrent.BlockingQueue;
/**
* @author hzk
* @date 2018/3/31
*/
public class Producer implements Runnable{
BlockingQueue blockingQueue;
public Producer(BlockingQueue blockingQueue) {
this.blockingQueue = blockingQueue;
}
@Override
public void run() {
try {
String threadIdentify = "A Producer,生产线程"+Thread.currentThread().getName();
blockingQueue.put(threadIdentify);
System.out.println("Produce success! Thread:"+Thread.currentThread().getName());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
Consumer
package com.ithzk.BlockingQueueTest;
import java.util.concurrent.BlockingQueue;
/**
* @author hzk
* @date 2018/3/31
*/
public class Consumer implements Runnable{
BlockingQueue blockingQueue;
public Consumer(BlockingQueue blockingQueue) {
this.blockingQueue = blockingQueue;
}
@Override
public void run() {
try {
String cOnsumer= Thread.currentThread().getName();
System.out.println("Current Consumer Thread:"+consumer);
//如果队列为空会阻塞当前线程
String take = blockingQueue.take();
System.out.println(consumer + " consumer get a product:"+take);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
BlockTest
package com.ithzk.BlockingQueueTest;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
/**
* @author hzk
* @date 2018/3/31
*/
public class BlockTest {
public static void main(String[] args) throws InterruptedException {
// 不设置的话，LinkedBlockingQueue默认大小为Integer.MAX_VALUE
// BlockingQueue blockingQueue = new LinkedBlockingQueue();
// BlockingQueue blockingQueue = new ArrayBlockingQueue(2);
BlockingQueue blockingQueue = new LinkedBlockingQueue(2);
Consumer cOnsumer= new Consumer(blockingQueue);
Producer producer = new Producer(blockingQueue);
for (int i = 0; i <3; i++) {
new Thread(producer, "Producer" + (i + 1)).start();
}
for (int i = 0; i <5; i++) {
new Thread(consumer, "Consumer" + (i + 1)).start();
}
Thread.sleep(5000);
new Thread(producer, "Producer" + (5)).start();
}
}
BlockingQueue有四个具体的实现类,常用的两种实现类为：
若不带大小参数,所生成的BlockingQueue的大小由Integer.MAX_VALUE来决定.其所含的对象是以FIFO(先入先出)顺序排序的。
LinkedBlockingQueue 可以指定容量，也可以不指定，不指定的话，默认最大是Integer.MAX_VALUE,其中主要用到put和take方法，put方法在队列满的时候会阻塞直到有队列成员被消费，take方法在队列空的时候会阻塞，直到有队列成员被放进来。
LinkedBlockingQueue和ArrayBlockingQueue区别
LinkedBlockingQueue和ArrayBlockingQueue比较起来,它们背后所用的数据结构不一样,导致LinkedBlockingQueue的数据吞吐量要大于ArrayBlockingQueue,但在线程数量很大时其性能的可预见性低于ArrayBlockingQueue.
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。
京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有