内容以及代码参考：Java高并发核心编程（卷2）：多线程、锁、JMM、JUC、高并发设计模式
场景：流程页面上，通过非标配置，明细表里的物料编码会自动根据RFC函数读取物料名称，描述，等级，采购组等具体数据
ThreadPoolExecutor线程池7大参数：
- corePoolSize： 核心线程数，一旦建立，不会因为超过存活时间而销毁，会一直存在复用。
- maximumPoolSize：最大线程数，核心线程数全部被使用了，会额外创建新线程供消费者使用，但有上限，上限就是这个最大线程数。
- keepAliveTime：存活时间，非核心线程，空闲时间一旦超过这个存活时间，会被销毁
- unit：时分秒
- workQueue：工作队列（阻塞队列）任务数超过最大线程数，会进入这个队列等待
- threadFactory：线程工厂，提供产生线程的静态工厂方法
- handler：拒绝策略，不填会默认
需要注意的是，corePoolSize这个参数，需要根据分类去设置大小，大概三种：CPU密集型、IO密集型、混合密集型。网上有公式，但具体要自己实测。
因为Java8以下的线程创建，直接对应的是电脑操作OS系统的物理线程的（底层是native方法），核心线程的创建、销毁、上下文切换，都涉及到物理操作，是需要消耗物理性能的。
另外，线程池的创建一般为单例，但是这是仅对同一个业务来说，不同的业务一般使用自己的线程池。
例如A业务用A单例线程池，B业务用B单例线程池。不用担心这会导致创建线程过多，线程的运行虽然对应的是CPU核心，但是不等于1个线程=占用一个核心。核心是一个公共计算资源，只有线程抢占到才能使用，所以线程是可以超过CPU核心数的。
还有IO类型的请求，其实并不占用太多的CPU计算。
ThreadPoolExecutor死锁问题：
根据网上定律，一个业务（即眼前这个读取SAP数据）应该配上一个独立的线程池。不能所有业务都使用一个线程池。
如果主业务和子业务都使用同一个线程池，可能会出现死锁阻塞情况。
如：主业务的任务数量填充了核心线程数，并且把最大线程数也占满了，但是主任务里面有子任务，而子任务也使用同一线程池，但是此刻线程池的数量全部被主业务占满了。所以子业务就会进入阻塞队列等待。两者互相冲突，就会造成死锁阻塞。
ThreadPoolExecutor临界资源处理的几个方向：
- 同步锁：synchronize、lock
- 线程隔离：threadlocal
- 原子类：AtomicInteger、AtomicBoolean等
其它待后续学习补充
代码：
ThreadUtil
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;
public class ThreadUtil {
/**
* CPU核数
**/
public static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
/**
* 空闲保活时限，单位秒
*/
public static final int KEEP_ALIVE_SECONDS = 30;
/**
* 有界队列size
*/
public static final int QUEUE_SIZE = 10000;
/**
* 混合线程池
*/
public static final int MIXED_CORE = 0; //混合线程池核心线程数
public static final int MIXED_MAX = 128; //最大线程数
public static final String MIXED_THREAD_AMOUNT = "mixed.thread.amount";
/**
* 核心线程数
*/
public static final int CORE_POOL_SIZE = 0;
public static final int MAXIMUM_POOL_SIZE = CPU_COUNT;
/**
* IO线程池最大线程数
*/
public static final int IO_MAX = Math.max(2, CPU_COUNT * 2);
/**
* IO线程池核心线程数
*/
public static final int IO_CORE = 0;
public static class CustomThreadFactory implements ThreadFactory {
//线程池数量
private static final AtomicInteger poolNumber = new AtomicInteger(1);
private final ThreadGroup group;
//线程数量
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final String threadTag;
public CustomThreadFactory(String threadTag) {
SecurityManager s = System.getSecurityManager();
group = (s != null) ? s.getThreadGroup() :
Thread.currentThread().getThreadGroup();
this.threadTag = "apppool-" + poolNumber.getAndIncrement() + "-" + threadTag + "-";
}
@Override
public Thread newThread(Runnable target) {
Thread t = new Thread(group, target,
threadTag + threadNumber.getAndIncrement(),
0);
if (t.isDaemon()) {
t.setDaemon(false);
}
if (t.getPriority() != Thread.NORM_PRIORITY) {
t.setPriority(Thread.NORM_PRIORITY);
}
return t;
}
}
/**
* 线程睡眠
*
* @param second 秒
*/
public static void sleepSeconds(int second) {
LockSupport.parkNanos(second * 1000L * 1000L * 1000L);
}
/**
* 线程睡眠
*
* @param millisecond 毫秒
*/
public static void sleepMilliSeconds(int millisecond) {
LockSupport.parkNanos(millisecond * 1000L * 1000L);
}
/**
* 获取当前线程名称
*/
public static String getCurThreadName() {
return Thread.currentThread().getName();
}
/**
* 获取当前线程ID
*/
public static long getCurThreadId() {
return Thread.currentThread().getId();
}
/**
* 获取当前线程
*/
public static Thread getCurThread() {
return Thread.currentThread();
}
/**
* 调用栈中的类名
*
* @return
*/
public static String stackClassName(int level) {
// Thread.currentThread().getStackTrace()[1]是当前方法 curClassName 执行堆栈
// Thread.currentThread().getStackTrace()[2]就是 curClassName 的 上一级的方法堆栈 以此类推
String className = Thread.currentThread().getStackTrace()[level].getClassName();//调用的类名
return className;
}
/**
* 调用栈中的方法名称
*
* @return
*/
public static String stackMethodName(int level) {
// Thread.currentThread().getStackTrace()[1]是当前方法 curMethodName 执行堆栈
// Thread.currentThread().getStackTrace()[2]就是 curMethodName 的 上一级的方法堆栈 以此类推
String className = Thread.currentThread().getStackTrace()[level].getMethodName();//调用的类名
return className;
}
public static void shutdownThreadPoolGracefully(ExecutorService threadPool) {
if (!(threadPool instanceof ExecutorService) || threadPool.isTerminated()) {
return;
}
try {
threadPool.shutdown(); //拒绝接受新任务
} catch (SecurityException e) {
return;
} catch (NullPointerException e) {
return;
}
try {
// 等待 60 s，等待线程池中的任务完成执行
if (!threadPool.awaitTermination(60, TimeUnit.SECONDS)) {
// 调用 shutdownNow 取消正在执行的任务
threadPool.shutdownNow();
// 再次等待 60 s，如果还未结束，可以再次尝试，或则直接放弃
if (!threadPool.awaitTermination(60, TimeUnit.SECONDS)) {
System.err.println("线程池任务未正常执行结束");
}
}
} catch (InterruptedException ie) {
// 捕获异常，重新调用 shutdownNow
threadPool.shutdownNow();
}
//任然没有关闭，循环关闭1000次，每次等待10毫秒
if (!threadPool.isTerminated()) {
try {
for (int i = 0; i < 1000; i++) {
if (threadPool.awaitTermination(10, TimeUnit.MILLISECONDS)) {
break;
}
threadPool.shutdownNow();
}
} catch (InterruptedException e) {
System.err.println(e.getMessage());
} catch (Throwable e) {
System.err.println(e.getMessage());
}
}
}
/**
* 获取执行IO密集型任务的线程池
*
* @return
*/
public static ThreadPoolExecutor getIoIntenseTargetThreadPool() {
return IoIntenseTargetThreadPoolLazyHolder.getInnerExecutor();
}
/**
* 获取执行CPU密集型任务的线程池
*
* @return
*/
public static ThreadPoolExecutor getCpuIntenseTargetThreadPool() {
return CpuIntenseTargetThreadPoolLazyHolder.getInnerExecutor();
}
/**
* 获取执行混合型任务的线程池 *
*
* @return
*/
public static ThreadPoolExecutor getMixedTargetThreadPool() {
return MixedTargetThreadPoolLazyHolder.getInnerExecutor();
}
}
ShutdownHookThread
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicInteger;
import com.engine.u9integration.util.LogUtil;
public class ShutdownHookThread extends Thread {
private volatile boolean hasShutdown = false;
private static AtomicInteger shutdownTimes = new AtomicInteger(0);
private final Callable callback;
/**
* Create the standard hook thread, with a call back, by using {@link Callable} interface.
*
* @param name
* @param callback The call back function.
*/
public ShutdownHookThread(String name, Callable callback) {
super("JVM退出钩子(" + name + ")");
this.callback = callback;
}
/**
* Thread run method.
* Invoke when the jvm shutdown.
*/
@Override
public void run() {
synchronized (this) {
// System.out.println(getName() + " starting.... ");
LogUtil.log("threadpool","threadpool").info(getName() + " starting.... ");
if (!this.hasShutdown) {
this.hasShutdown = true;
long beginTime = System.currentTimeMillis();
try {
this.callback.call();
} catch (Exception e) {
// System.out.println(getName() + " error: " + e.getMessage());
LogUtil.log("threadpool","threadpool").info(getName() + " error: " + e.getMessage());
}
long consumingTimeTotal = System.currentTimeMillis() - beginTime;
// System.out.println(getName() + " 耗时(ms): " + consumingTimeTotal);
LogUtil.log("threadpool","threadpool").info(getName() + " 耗时(ms): " + consumingTimeTotal);
}
}
}
}
IoIntenseTargetThreadPoolLazyHolder
import java.util.concurrent.Callable;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import static com.engine.util.ThreadUtil.*;
public class IoIntenseTargetThreadPoolLazyHolder {
//线程池： 用于IO密集型任务
private static final ThreadPoolExecutor EXECUTOR = new ThreadPoolExecutor(
IO_CORE,
IO_MAX,
KEEP_ALIVE_SECONDS,
TimeUnit.SECONDS,
new LinkedBlockingQueue(QUEUE_SIZE),
new ThreadUtil.CustomThreadFactory("io"));
public static ThreadPoolExecutor getInnerExecutor() {
return EXECUTOR;
}
static {
// log.info("线程池已经初始化");
LogUtil.log("threadpool","threadpool").info("线程池已经初始化");
LogUtil.log("threadpool","threadpool").info("IO_MAX:"+IO_MAX);
EXECUTOR.allowCoreThreadTimeOut(true);
//JVM关闭时的钩子函数
Runtime.getRuntime().addShutdownHook(
new ShutdownHookThread("IO密集型任务线程池", new Callable<Void>() {
@Override
public Void call() throws Exception {
//优雅关闭线程池
shutdownThreadPoolGracefully(EXECUTOR);
return null;
}
}));
}
}
CpuIntenseTargetThreadPoolLazyHolder
import java.util.concurrent.Callable;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import static com.engine.util.ThreadUtil.*;
public class CpuIntenseTargetThreadPoolLazyHolder {
//线程池： 用于CPU密集型任务
private static final ThreadPoolExecutor EXECUTOR = new ThreadPoolExecutor(
MAXIMUM_POOL_SIZE,
MAXIMUM_POOL_SIZE,
KEEP_ALIVE_SECONDS,
TimeUnit.SECONDS,
new LinkedBlockingQueue(QUEUE_SIZE),
new CustomThreadFactory("cpu"));
public static ThreadPoolExecutor getInnerExecutor() {
return EXECUTOR;
}
static {
// log.info("线程池已经初始化");
LogUtil.log("threadpool","threadpool").info("线程池已经初始化");
EXECUTOR.allowCoreThreadTimeOut(true);
//JVM关闭时的钩子函数
Runtime.getRuntime().addShutdownHook(
new ShutdownHookThread("IO密集型任务线程池", new Callable<Void>() {
@Override
public Void call() throws Exception {
//优雅关闭线程池
shutdownThreadPoolGracefully(EXECUTOR);
return null;
}
}));
}
}
MixedTargetThreadPoolLazyHolder
import java.util.concurrent.Callable;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import static com.engine.util.ThreadUtil.*;
public class MixedTargetThreadPoolLazyHolder {
//首先从环境变量 mixed.thread.amount 中获取预先配置的线程数
//如果没有对 mixed.thread.amount 做配置，则使用常量 MIXED_MAX 作为线程数
private static final int max = (null != System.getProperty(MIXED_THREAD_AMOUNT)) ?
Integer.parseInt(System.getProperty(MIXED_THREAD_AMOUNT)) : MIXED_MAX;
//线程池： 用于混合型任务
private static final ThreadPoolExecutor EXECUTOR = new ThreadPoolExecutor(
max,
max,
KEEP_ALIVE_SECONDS,
TimeUnit.SECONDS,
new LinkedBlockingQueue(QUEUE_SIZE),
new CustomThreadFactory("mixed"));
public static ThreadPoolExecutor getInnerExecutor() {
return EXECUTOR;
}
static {
// log.info("线程池已经初始化");
LogUtil.log("threadpool","threadpool").info("线程池已经初始化");
EXECUTOR.allowCoreThreadTimeOut(true);
//JVM关闭时的钩子函数
Runtime.getRuntime().addShutdownHook(new ShutdownHookThread("混合型任务线程池", new Callable<Void>() {
@Override
public Void call() throws Exception {
//优雅关闭线程池
shutdownThreadPoolGracefully(EXECUTOR);
return null;
}
}));
}
}
YclblclImportToSap2
package com.engine.service.impl;
import com.engine.service.BaseAfter;
import com.engine.util.SapRfc;
import com.engine.util.SapUtil;
import com.engine.util.ThreadUtil;
import com.engine.u9integration.util.LogUtil;
import com.weaverboot.frame.ioc.anno.classAnno.WeaIocReplaceComponent;
import com.weaverboot.frame.ioc.anno.methodAnno.WeaReplaceAfter;
import com.weaverboot.frame.ioc.handler.replace.weaReplaceParam.impl.WeaAfterReplaceParam;
import weaver.conn.RecordSet;
import weaver.general.BaseBean;
import weaver.general.StringUtil;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
/**
* 成本中心领料(杂发)批量导入根据物料、存储地点、批次编号获取SAP唯一数据
* 自动填充表单物料描述、规格等字段
*/
@WeaIocReplaceComponent("yclblclImportToSap") //如不标注名称，则按类的全路径注入
public class YclblclImportToSap2 extends BaseBean implements BaseAfter {
//这个是接口后置方法，大概的用法跟前置方法差不多，稍有差别
//注解名称为WeaReplaceAfter
//返回类型必须为String
//参数叫WeaAfterReplaceParam，这个类前四个参数跟前置方法的那个相同，不同的是多了一个叫data的String，这个是那个接口执行完返回的报文
//你可以对那个报文进行操作，然后在这个方法里return回去
@WeaReplaceAfter(value = "/api/workflow/reqform/doImportDetail",order = 6)
@Override
public String after(WeaAfterReplaceParam weaAfterReplaceParam) {
long l1 = System.currentTimeMillis();
/*
用于确定是否是成本中心领料(杂发)的导入动作。
测试环境：SELECT * from workflow_base where formid = -291 对应表：formtable_main_291
正式环境：SELECT * from workflow_base where formid = -506 对应表：formtable_main_506
*/
String data = weaAfterReplaceParam.getData();//这个就是接口执行完的报文
String formid = weaAfterReplaceParam.getParamMap().get("formid").toString();
String requestid = weaAfterReplaceParam.getParamMap().get("requestid").toString();
//成本中心领料(测试中)-formid
String yclblclFormId = getPropValue("XX","yclblclFormId");
//成本中心领料(测试中)-表名
String yclblclTableName = getPropValue("XX","yclblclTableName");
//writeLog("cczxllFormId:"+cczxllFormId);
// LogUtil.log("sap","sap").info("yclblclFormId:"+yclblclFormId.toString());
// LogUtil.log("sap","sap").info("yclblclTableName:"+yclblclTableName.toString());
// LogUtil.log("sap","sap").info("formid:"+formid.toString());
if (!formid.equals(yclblclFormId)){
return data;
}
RecordSet rs = new RecordSet();
//根据物料编号、工厂、存储地点、批次编号其中三个以上条件可以在SAP查出唯一的物料库存。当然，如果不唯一，直接废弃，让用户自己去选择
String gc = getMainDataByRequestid(requestid,yclblclTableName,2);
String mainId = getMainDataByRequestid(requestid,yclblclTableName,1);
// LogUtil.log("sap","sap").info("gc:"+gc.toString());
// LogUtil.log("sap","sap").info("mainId:"+mainId.toString());
String sql2 = "select id,pc,wlbh from "+yclblclTableName+"_dt1 where mainid = "+mainId;
// LogUtil.log("sap","sap").info("sql2:"+sql2.toString());
rs.execute(sql2);
int counts = rs.getCounts();
if (counts>200) {
return data;
}
//倒数闩，需要倒数counts次
CountDownLatch latch = new CountDownLatch(counts);
while (rs.next()){
String wlbh = rs.getString("wlbh");
String ccdd = rs.getString("dccbm");
String pc = rs.getString("pc");
String detailId = rs.getString("id");
//io密集型线程池处理
ThreadUtil.getIoIntenseTargetThreadPool().submit(()->{updateOaDetailBySapData( wlbh, gc, ccdd,
pc, detailId, yclblclTableName,latch);});
}
try {
latch.await(); // 等待倒数闩的次数减少到0，所有的线程执行完成
} catch (InterruptedException e) {
e.printStackTrace();
}
long l2 = System.currentTimeMillis();
long l = l2 - l1;
LogUtil.log("sap","sap").info("currentTimeMillis:"+l);
return data;
}
/**
* 如果读取SAP函数有数据返回，就拿第一条数据updateOA的明细表
* @param wlbh 物料编码
* @param gc 工厂
* @param ccdd 存储地点
* @param pc 批次
* @param detailId 明细表ID
* @param tableName 明细表表名
*/
private void updateOaDetailBySapData(String wlbh,String gc,String ccdd,
String pc,String detailId,String tableName,CountDownLatch latch){
try {
LogUtil.log("sap","sap").info("updateOaDetailBySapData222:");
SapRfc zmm_fm_002 = new SapRfc("ZMM_FM_002");
if (!StringUtil.isEmpty(wlbh)){
zmm_fm_002.setImportValue("IV_MATNR",wlbh);
}
if (!StringUtil.isEmpty(gc)){
zmm_fm_002.setImportValue("IV_WEAKS",gc);
}
if (!StringUtil.isEmpty(ccdd)){
zmm_fm_002.setImportValue("IV_LGORT",ccdd);
}
if (!StringUtil.isEmpty(pc)){
zmm_fm_002.setImportValue("IV_CHARG",pc);
}
zmm_fm_002.execute();
int it_base = zmm_fm_002.getExportTableCount("IT_DATA");
LogUtil.log("sap","sap").info("it_base:"+it_base);
if (it_base>0){
Set<String> set = new HashSet<>();
set.add("MAKTX");
set.add("LABST");
set.add("MATNR");
set.add("MEINS");
//set.add("LGORT"); 存储地点
set.add("CHARG");
set.add("ZCHGNO");
set.add("GROES");
set.add("ZZXH");
set.add("ZLENGTH");
set.add("ZTHICK");
set.add("ZWIDTH");
set.add("ZANGEL");
set.add("ZPZZK");
HashMap it_data1 = zmm_fm_002.getExportTable("IT_DATA", 0, set);
//执行完清空client
// SapUtil.releaseClient2(zmm_fm_002);
StringBuffer sb = getUpdateSql(it_data1);
String sql3 = "update "+tableName+"_dt1 set "+sb.toString()+" where id = "+detailId;
RecordSet rs2 = new RecordSet();
// LogUtil.log("sap","sap").info("updateOaDetailBySapData sql:"+sql3.toString());
rs2.execute(sql3);
}
} catch (Exception e) {
e.printStackTrace();
}finally {
latch.countDown();// 倒数闩减少一次
}
}
/**
* 根据requestid获取main的数据，1返回mainId，2返回工厂
* 有缓存，不怕多次查询
* @param requestid
* @param tableName
* @return
*/
private String getMainDataByRequestid(String requestid,String tableName,int statu){
RecordSet rs = new RecordSet();
String sql = "select gc,id from "+tableName+" where requestid = "+requestid;
rs.execute(sql);
String mainId = null;
if (rs.next()){
if (statu==1){
mainId = rs.getString("id");
}
if (statu==2){
mainId = rs.getString("gc");
}
}
return mainId;
}
/**
* 将map中要update的字段转化为SQL语句片段
* @param map
* @return
*/
private StringBuffer getUpdateSql(Map map){
StringBuffer sb = new StringBuffer();
map.forEach((key,value)->{
sb.append(getFields(key.toString()));
sb.append("=");
sb.append("'"+value+"',");
});
sb.deleteCharAt(sb.length()-1);
return sb;
}
private String getFields(String sapFields){
String oaFields = "";
switch(sapFields) {
case "MAKTX":
oaFields = "wlms";
break;
case "LABST":
oaFields = "kcsl";
break;
case "MATNR":
oaFields = "wlbh";
break;
case "MEINS":
oaFields = "dw";
break;
// case "LGORT":
// oaFields = "ccdd";
// break;
case "CHARG":
oaFields = "pc";
break;
case "ZCHGNO":
oaFields = "ph";
break;
case "GROES":
oaFields = "gg";
break;
case "ZZXH":
oaFields = "xh";
break;
case "ZLENGTH":
oaFields = "c";
break;
case "ZWIDTH":
oaFields = "k";
break;
case "ZTHICK":
oaFields = "hd";
break;
case "ZANGEL":
oaFields = "jd";
break;
case "ZPZZK":
oaFields = "pzzk";
break;
}
return oaFields;
}
}
标签：String,明细,util,static,线程,import,SAP,public From： https://www.cnblogs.com/stfzhuang/p/17921511.html