Skip to content

D
dp
Commit fb93c651 authored by Spencer Chang's avatar Spencer Chang
Browse files
Options

[fix]

parent d078dbc3
Show whitespace changes
Inline Side-by-side
Showing with 181 additions and 104 deletions
v2-多线程版本/java-code/dp/controllers/SysDpExecuteHistoryController.java
View file @ fb93c651
......@@ -44,7 +44,7 @@ public class SysDpExecuteHistoryController extends BaseController {
final String param = (String) requestData.get("parameter");
final String dpContext = URLDecoder.decode(new String(Base64.decodeBase64(param), StandardCharsets.UTF_8),"UTF-8");
if (ck == 0){
service.query4ck(requestContext, ip, dpContext);
service.query4Verify(requestContext, ip, dpContext);
}else if(ck == 1){
request.setAttribute("_NO_PAGE_","Y");
final String context = new String(dpContext.getBytes(StandardCharsets.ISO_8859_1), StandardCharsets.UTF_8);
......
v2-多线程版本/java-code/dp/core/SubExecSqlContext.java → v2-多线程版本/java-code/dp/core/ExecSqlContext.java
View file @ fb93c651
......@@ -3,12 +3,14 @@ package com.hand.hls.dp.core;
import com.hand.hap.core.IRequest;
import com.hand.hls.dp.service.SysDpExecuteHistoryService;
import java.util.Objects;
/**
* @author <a href="mailto:zhangnet14@gmail.com">Spencer Chang</a>
* @date 2020/8/13 09:50
* @since 1.0
* sql执行上下文
*/
public class SubExecSqlContext {
public class ExecSqlContext {
private SysDpExecuteHistoryService service;
private IRequest requestContext;
private String sql;
......@@ -63,4 +65,29 @@ public class SubExecSqlContext {
public void setSeq(Integer seq) {
this.seq = seq;
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
ExecSqlContext that = (ExecSqlContext) o;
return Objects.equals(service, that.service) && Objects.equals(requestContext, that.requestContext) && Objects.equals(sql, that.sql) && Objects.equals(ip, that.ip) && Objects.equals(execType, that.execType) && Objects.equals(seq, that.seq);
}
@Override
public int hashCode() {
return Objects.hash(service, requestContext, sql, ip, execType, seq);
}
@Override
public String toString() {
return "ExecSqlContext{" +
"service=" + service +
", requestContext=" + requestContext +
", sql='" + sql + '\'' +
", ip='" + ip + '\'' +
", execType='" + execType + '\'' +
", seq=" + seq +
'}';
}
}
v2-多线程版本/java-code/dp/service/SysDpExecuteHistoryService.java
View file @ fb93c651
......@@ -14,7 +14,7 @@ import javax.servlet.http.HttpServletResponse;
* @since
*/
public interface SysDpExecuteHistoryService extends IBaseService<SysDpExecuteHistory>, ProxySelf<SysDpExecuteHistoryService> {
void query4ck(IRequest requestContext, final String ip, final String dpContext) throws Exception;
void query4Verify(IRequest requestContext, final String ip, final String dpContext) throws Exception;
void query(IRequest requestContext, HttpServletResponse response, final String ip, final String dpContext) throws Exception;
ResponseData insert(IRequest requestContext, final String ip, final String dpContext);
ResponseData update(IRequest requestContext, final String ip, final String dpContext);
......
v2-多线程版本/java-code/dp/service/impl/SysClusterLogFileSubscriber.java
View file @ fb93c651
......@@ -2,6 +2,7 @@ package com.hand.hls.dp.service.impl;
import com.hand.hap.message.IMessageConsumer;
import com.hand.hap.message.TopicMonitor;
import com.hand.hls.dp.util.IPUtils;
import com.hand.hls.dp.util.LogUtils;
import org.apache.commons.codec.binary.Base64;
import org.slf4j.Logger;
......@@ -48,14 +49,14 @@ public class SysClusterLogFileSubscriber implements IMessageConsumer<String>, In
ZipOutputStream zos = new ZipOutputStream(cos);
// 当天日志名称
String currLogFileName = logPath + "leaf.log";
String localServerIp = IPUtils.getIpAddressLocal();
// 查找日志目录下所有文件
List<String> files = new ArrayList<>(31);
LogUtils.findFiles(new File(logPath), files);
files.forEach(file -> {
if (file.equals(currLogFileName)) {
// 创建ZipEntry
ZipEntry entry = new ZipEntry(file.substring(logPath.length()));
ZipEntry entry = new ZipEntry(localServerIp+"-"+file.substring(logPath.length()));
// 放入文件，防止文件名相同的附件报错
try {
zos.putNextEntry(entry);
......@@ -70,20 +71,21 @@ public class SysClusterLogFileSubscriber implements IMessageConsumer<String>, In
zos.closeEntry();
zos.close();
if (logger.isDebugEnabled()) {
logger.debug("redis key: [{}]", LOGFILE_KEY);
logger.debug("[{}] redis key: [{}]", localServerIp, LOGFILE_KEY);
}
redisTemplate.opsForList().rightPush(LOGFILE_KEY, Base64.encodeBase64String(byteOutPut.toByteArray()));
cos.close();
} catch (Exception e) {
logger.warn(e.getMessage());
logger.warn("[{}] 日志暂存redis 出错: [{}]", localServerIp, e.getMessage());
}
}
@Override
public void onMessage(String message, String pattern) {
if (LOGFILE_DOWNLOAD.equals(pattern)) {
String localServerIp = IPUtils.getIpAddressLocal();
if (logger.isDebugEnabled()) {
logger.debug("start do [{}]", message);
logger.debug("[{}] start do [{}]", localServerIp, message);
}
// 收集当天日志暂存redis
collectLog();
......
v2-多线程版本/java-code/dp/service/impl/SysDpDownloadLogServiceImpl.java
View file @ fb93c651
......@@ -19,6 +19,7 @@ import java.io.ByteArrayOutputStream;
import java.io.File;
import java.nio.charset.StandardCharsets;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Date;
......@@ -66,9 +67,9 @@ public class SysDpDownloadLogServiceImpl implements SysDpDownloadLogService {
*/
@Override
public void downloadClusterLogZipFile(IRequest requestContext, HttpServletRequest request, HttpServletResponse response) throws Exception {
LocalDate tmp = LocalDate.now();
Date date = Date.from(tmp.atStartOfDay().atZone(ZoneId.systemDefault()).toInstant());
String zipFileName = "leaf-" + DateUtil.format(date, "yyyy-MM-dd") + ".zip";
LocalDateTime tmp = LocalDateTime.now();
Date date = Date.from(tmp.atZone(ZoneId.systemDefault()).toInstant());
String zipFileName = "leaf-cluster-" + DateUtil.format(date, "yyyyMMddHHmmss") + ".zip";
response.reset();
response.setHeader("Pragma", "No-cache");
response.setHeader("Cache-Control", "no-cache,must-revalidate");
......@@ -103,11 +104,12 @@ public class SysDpDownloadLogServiceImpl implements SysDpDownloadLogService {
byte[] buffer = new byte[LogUtils.DEFAULT_LARGE_BUFFER_SIZE];
while (LogUtils.EOF != (n = byteArrayInputStream.read(buffer))) {
zos.write(buffer, 0, n);
zos.flush();
}
}
zos.closeEntry();
zos.close();
response.setHeader("Content-Length", String.valueOf(byteOutPut.toByteArray().length));
response.setHeader("Content-Length", String.valueOf(byteOutPut.size()));
IOUtils.copy(new ByteArrayInputStream(byteOutPut.toByteArray()), response.getOutputStream());
cos.close();
}
......
v2-多线程版本/java-code/dp/service/impl/SysDpExecuteHistoryServiceImpl.java
View file @ fb93c651
......@@ -5,20 +5,23 @@ import com.alibaba.druid.sql.parser.ParserException;
import com.alibaba.druid.util.JdbcConstants;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.ttl.TransmittableThreadLocal;
import com.alibaba.ttl.threadpool.TtlExecutors;
import com.hand.hap.core.IRequest;
import com.hand.hap.generator.service.impl.DBUtil;
import com.hand.hap.system.dto.ResponseData;
import com.hand.hap.system.service.impl.BaseServiceImpl;
import com.hand.hls.dp.core.SubExecSqlContext;
import com.hand.hls.dp.core.ExecSqlContext;
import com.hand.hls.dp.dto.SysDpExecuteHistory;
import com.hand.hls.dp.service.SysDpExecuteHistoryService;
import com.hand.hls.dp.util.Maps;
import com.hand.hls.dp.util.SqlConstantUtils;
import com.hand.hls.dp.util.SqlParserCheckUtils;
import com.hand.hls.dp.util.SqlCheckUtils;
import com.hand.hls.exception.HlsCusException;
import hls.core.sys.mapper.SysCodeValueMapper;
import leaf.plugin.export.components.ExcelExportUtil;
import leaf.utils.ConfigUtils;
import leaf.util.ConfigUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
......@@ -42,12 +45,13 @@ import java.sql.SQLException;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.ReentrantLock;
/**
......@@ -79,13 +83,13 @@ public class SysDpExecuteHistoryServiceImpl extends BaseServiceImpl<SysDpExecute
* @param dpContext 待执行内容
*/
@Override
public void query4ck(IRequest requestContext, String ip, String dpContext) {
public void query4Verify(IRequest requestContext, String ip, String dpContext) {
Objects.requireNonNull(dpContext, "传入的语句为空！");
if (logger.isInfoEnabled()) {
logger.info("==> start execute query4ck");
logger.info("ip:[{}] ==> 检验查询待执行语句:[{}]", ip, dpContext);
}
final String ipWhiteCheck = sysCodeValueMapper.queryCodeValueMeaning("IP_ALLOW_LIST", "IP");
final String ipAllowVerify = sysCodeValueMapper.queryCodeValueMeaning("IP_ALLOW_LIST", "IP");
if (StringUtils.isNotBlank(dpContext)) {
StringBuilder msg = new StringBuilder();
......@@ -108,7 +112,7 @@ public class SysDpExecuteHistoryServiceImpl extends BaseServiceImpl<SysDpExecute
}
String checkMsg = null;
// 非内部限制
if (!StringUtils.startsWith(ip, ipWhiteCheck)) {
if (!StringUtils.startsWith(ip, ipAllowVerify)) {
checkMsg = SqlParserCheckUtils.parserCheckSelect(formatSql, true);
} else {
checkMsg = SqlParserCheckUtils.parserCheckSelect(formatSql, false);
......@@ -173,7 +177,7 @@ public class SysDpExecuteHistoryServiceImpl extends BaseServiceImpl<SysDpExecute
logger.info("==> start execute query");
logger.info("ip:[{}] ==> 查询待执行语句:[{}]", ip, dpContext);
}
final String ipWhiteCheck = sysCodeValueMapper.queryCodeValueMeaning("IP_ALLOW_LIST", "IP");
final String ipAllowVerify = sysCodeValueMapper.queryCodeValueMeaning("IP_ALLOW_LIST", "IP");
SysDpExecuteHistory sysDpExecuteHistory = new SysDpExecuteHistory();
sysDpExecuteHistory.setDehIp(ip);
sysDpExecuteHistory.setDehUserId(requestContext.getUserId());
......@@ -204,7 +208,7 @@ public class SysDpExecuteHistoryServiceImpl extends BaseServiceImpl<SysDpExecute
}
String checkMsg = null;
// 非内部限制
if (!StringUtils.startsWith(ip, ipWhiteCheck)) {
if (!StringUtils.startsWith(ip, ipAllowVerify)) {
checkMsg = SqlParserCheckUtils.parserCheckSelect(formatSql, true);
} else {
checkMsg = SqlParserCheckUtils.parserCheckSelect(formatSql, false);
......@@ -229,7 +233,7 @@ public class SysDpExecuteHistoryServiceImpl extends BaseServiceImpl<SysDpExecute
// 组装excel列信息
for (int i = 1; i <= columnCount; i++) {
String columnName = md.getColumnName(i);
Map<String, Object> columnInfo = new HashMap<>(4);
Map<String, Object> columnInfo = Maps.newHashMapWithExpectedSize(4);
columnInfo.put("width", 150);
columnInfo.put("name", columnName);
columnInfo.put("align", "left");
......@@ -238,7 +242,7 @@ public class SysDpExecuteHistoryServiceImpl extends BaseServiceImpl<SysDpExecute
columnConfig.add(columnJsonObject);
}
while (rs.next()) {
Map<String, Object> rowData = new HashMap<>(1);
Map<String, Object> rowData = Maps.newHashMapWithExpectedSize(1);
try {
getType(rs, md, rowData);
} catch (Exception e) {
......@@ -261,10 +265,12 @@ public class SysDpExecuteHistoryServiceImpl extends BaseServiceImpl<SysDpExecute
}
self().insertSelective(requestContext, sysDpExecuteHistory);
// 查询结果导出excel
long startTime = System.currentTimeMillis();
ExcelExportUtil excelExportUtil = new ExcelExportUtil();
excelExportUtil.createExcel(response, columnConfig, retList);
long endTime = System.currentTimeMillis();
if (logger.isInfoEnabled()) {
logger.info("<== end execute query");
logger.info("<== end execute query 耗时：[{}]", endTime - startTime);
}
}
......@@ -283,7 +289,7 @@ public class SysDpExecuteHistoryServiceImpl extends BaseServiceImpl<SysDpExecute
logger.info("==> start execute insert");
logger.info("ip:[{}] ==> 新增待执行语句:[{}]", ip, dpContext);
}
final String ipWhiteCheck = sysCodeValueMapper.queryCodeValueMeaning("IP_ALLOW_LIST", "IP");
final String ipAllowVerify = sysCodeValueMapper.queryCodeValueMeaning("IP_ALLOW_LIST", "IP");
ResponseData responseData = new ResponseData(false);
SysDpExecuteHistory sysDpExecuteHistory = new SysDpExecuteHistory();
sysDpExecuteHistory.setDehIp(ip);
......@@ -305,7 +311,7 @@ public class SysDpExecuteHistoryServiceImpl extends BaseServiceImpl<SysDpExecute
insertDpExecuteHistory.setDehStatus(SqlConstantUtils.SQL_EXEC_UNDO);
// 非内部限制
if (!StringUtils.startsWith(ip, ipWhiteCheck)) {
if (!StringUtils.startsWith(ip, ipAllowVerify)) {
if (SqlCheckUtils.checkSqlFilterTableIgnoreCase(dpContext, SqlConstantUtils.SQL_FILTER_TABLE)) {
msg.append("当前网络环境不允许新增数据！");
}
......@@ -379,7 +385,7 @@ public class SysDpExecuteHistoryServiceImpl extends BaseServiceImpl<SysDpExecute
logger.info("==> start execute update");
logger.info("ip:[{}] ==> 更新待执行语句:[{}]", ip, dpContext);
}
final String ipWhiteCheck = sysCodeValueMapper.queryCodeValueMeaning("IP_ALLOW_LIST", "IP");
final String ipAllowVerify = sysCodeValueMapper.queryCodeValueMeaning("IP_ALLOW_LIST", "IP");
ResponseData responseData = new ResponseData(false);
SysDpExecuteHistory sysDpExecuteHistory = new SysDpExecuteHistory();
sysDpExecuteHistory.setDehIp(ip);
......@@ -401,7 +407,7 @@ public class SysDpExecuteHistoryServiceImpl extends BaseServiceImpl<SysDpExecute
updateDpExecuteHistory.setDehStatus(SqlConstantUtils.SQL_EXEC_UNDO);
// 非内部限制
if (!StringUtils.startsWith(ip, ipWhiteCheck)) {
if (!StringUtils.startsWith(ip, ipAllowVerify)) {
if (SqlCheckUtils.checkSqlFilterTableIgnoreCase(dpContext, SqlConstantUtils.SQL_FILTER_TABLE)) {
msg.append("当前网络环境不允许修改数据！");
}
......@@ -544,7 +550,7 @@ public class SysDpExecuteHistoryServiceImpl extends BaseServiceImpl<SysDpExecute
logger.info("==> start execute delete");
logger.info("ip:[{}] ==> 删除待执行语句:[{}]", ip, dpContext);
}
final String ipWhiteCheck = sysCodeValueMapper.queryCodeValueMeaning("IP_ALLOW_LIST", "IP");
final String ipAllowVerify = sysCodeValueMapper.queryCodeValueMeaning("IP_ALLOW_LIST", "IP");
ResponseData responseData = new ResponseData(false);
SysDpExecuteHistory sysDpExecuteHistory = new SysDpExecuteHistory();
sysDpExecuteHistory.setDehIp(ip);
......@@ -566,7 +572,7 @@ public class SysDpExecuteHistoryServiceImpl extends BaseServiceImpl<SysDpExecute
delDpExecuteHistory.setDehStatus(SqlConstantUtils.SQL_EXEC_UNDO);
// 非内部限制
if (!StringUtils.startsWith(ip, ipWhiteCheck)) {
if (!StringUtils.startsWith(ip, ipAllowVerify)) {
if (SqlCheckUtils.checkSqlFilterTableIgnoreCase(dpContext, SqlConstantUtils.SQL_FILTER_TABLE)) {
msg.append("当前网络环境不允许删除数据！");
}
......@@ -697,7 +703,7 @@ public class SysDpExecuteHistoryServiceImpl extends BaseServiceImpl<SysDpExecute
logger.info("==> start execute ddl");
logger.info("ip:[{}] ==> 待执行DDL语句:[{}]", ip, dpContext);
}
final String ipWhiteCheck = sysCodeValueMapper.queryCodeValueMeaning("IP_ALLOW_LIST", "IP");
final String ipAllowVerify = sysCodeValueMapper.queryCodeValueMeaning("IP_ALLOW_LIST", "IP");
ResponseData responseData = new ResponseData(false);
SysDpExecuteHistory sysDpExecuteHistory = new SysDpExecuteHistory();
sysDpExecuteHistory.setDehIp(ip);
......@@ -709,7 +715,7 @@ public class SysDpExecuteHistoryServiceImpl extends BaseServiceImpl<SysDpExecute
if (StringUtils.isNotBlank(dpContext)) {
StringBuilder msg = new StringBuilder();
// 非内部限制
if (!StringUtils.startsWith(ip, ipWhiteCheck)) {
if (!StringUtils.startsWith(ip, ipAllowVerify)) {
msg.append("当前网络环境不允许执行语句！");
}
......@@ -918,13 +924,11 @@ public class SysDpExecuteHistoryServiceImpl extends BaseServiceImpl<SysDpExecute
if (length <= 0) {
throw new HlsCusException("没有可执行语句！");
}
ConcurrentHashMap<Integer, String> retMsg = new ConcurrentHashMap<>(length);
ConcurrentHashMap<Integer, String> retMsg = new ConcurrentHashMap<>(length,0.75f);
List<String> result = new ArrayList<>(retMsg.size());
CountDownLatch downLatch = new CountDownLatch(length);
SubExecSqlContext subExecContext = new SubExecSqlContext();
subExecContext.setService(self());
subExecContext.setRequestContext(requestContext);
subExecContext.setIp(ip);
ExecutorService service = Executors.newFixedThreadPool(10);
ExecutorService executorService = TtlExecutors.getTtlExecutorService(service);
for (int i = 1; i <= length; i++) {
String undo = undos[i - 1];
if (StringUtils.isBlank(undo)) {
......@@ -948,8 +952,12 @@ public class SysDpExecuteHistoryServiceImpl extends BaseServiceImpl<SysDpExecute
if (StringUtils.isNotEmpty(msg.toString())) {
throw new HlsCusException(msg.toString());
}
subExecContext.setSeq(i);
subExecContext.setSql(formatSql);
ExecSqlContext subExecSqlContext = new ExecSqlContext();
subExecSqlContext.setService(self());
subExecSqlContext.setRequestContext(requestContext);
subExecSqlContext.setIp(ip);
subExecSqlContext.setSeq(i);
subExecSqlContext.setSql(formatSql);
// 判断语句开始关键字
if (StringUtils.startsWithIgnoreCase(formatSql, SqlConstantUtils.SQL_EXEC_SELECT)) {
throw new HlsCusException("执行按钮不支持SELECT查询语句，请使用查询按钮");
......@@ -959,39 +967,47 @@ public class SysDpExecuteHistoryServiceImpl extends BaseServiceImpl<SysDpExecute
if (StringUtils.isNotEmpty(msg.toString())) {
throw new HlsCusException(msg.toString());
}
subExecContext.setExecType(SqlConstantUtils.SQL_EXEC_INSERT);
taskExecutor.execute(new SubThreadTask(subExecContext, retMsg, downLatch));
subExecSqlContext.setExecType(SqlConstantUtils.SQL_EXEC_INSERT);
TransmittableThreadLocal<ExecSqlContext> ttl = new TransmittableThreadLocal<>();
ttl.set(subExecSqlContext);
executorService.execute(new SubThreadTask(ttl, retMsg, downLatch));
} else if (StringUtils.startsWithIgnoreCase(formatSql, SqlConstantUtils.SQL_EXEC_UPDATE)) {
String checkMsg = SqlParserCheckUtils.parserCheckUpdate(formatSql);
msg.append(checkMsg);
if (StringUtils.isNotEmpty(msg.toString())) {
throw new HlsCusException(msg.toString());
}
subExecContext.setExecType(SqlConstantUtils.SQL_EXEC_UPDATE);
taskExecutor.execute(new SubThreadTask(subExecContext, retMsg, downLatch));
subExecSqlContext.setExecType(SqlConstantUtils.SQL_EXEC_UPDATE);
TransmittableThreadLocal<ExecSqlContext> ttl = new TransmittableThreadLocal<>();
ttl.set(subExecSqlContext);
executorService.execute(new SubThreadTask(ttl, retMsg, downLatch));
} else if (StringUtils.startsWithIgnoreCase(formatSql, SqlConstantUtils.SQL_EXEC_DELETE)) {
String checkMsg = SqlParserCheckUtils.parserCheckDelete(formatSql);
msg.append(checkMsg);
if (StringUtils.isNotEmpty(msg.toString())) {
throw new HlsCusException(msg.toString());
}
subExecContext.setExecType(SqlConstantUtils.SQL_EXEC_DELETE);
taskExecutor.execute(new SubThreadTask(subExecContext, retMsg, downLatch));
subExecSqlContext.setExecType(SqlConstantUtils.SQL_EXEC_DELETE);
TransmittableThreadLocal<ExecSqlContext> ttl = new TransmittableThreadLocal<>();
ttl.set(subExecSqlContext);
executorService.execute(new SubThreadTask(ttl, retMsg, downLatch));
} else if (StringUtils.startsWithIgnoreCase(formatSql, SqlConstantUtils.DDL_CREATE_TABLE) ||
StringUtils.startsWithIgnoreCase(formatSql, SqlConstantUtils.DDL_ALTER_TABLE) ||
StringUtils.startsWithIgnoreCase(formatSql, SqlConstantUtils.DDL_COMMENT_TABLE) ||
StringUtils.startsWithIgnoreCase(formatSql, SqlConstantUtils.DDL_COMMENT_ON_COLUMN) ||
StringUtils.startsWithIgnoreCase(formatSql, SqlConstantUtils.DDL_CREATE_INDEX) ||
StringUtils.startsWithIgnoreCase(formatSql, SqlConstantUtils.DDL_CREATE_SEQUENCE) ||
StringUtils.startsWithIgnoreCase(formatSql, SqlConstantUtils.DDL_CREATE_SEQUENCE) ||
StringUtils.startsWithIgnoreCase(formatSql, SqlConstantUtils.DDL_ALTER_SEQUENCE) ||
StringUtils.startsWithIgnoreCase(formatSql, SqlConstantUtils.DDL_ALTER_SYSTEM_KILL_SESSION)) {
String checkMsg = SqlParserCheckUtils.parserCheckDDL(formatSql);
msg.append(checkMsg);
if (StringUtils.isNotEmpty(msg.toString())) {
throw new HlsCusException(msg.toString());
}
subExecContext.setExecType(SqlConstantUtils.SQL_EXEC_DDL);
taskExecutor.execute(new SubThreadTask(subExecContext, retMsg, downLatch));
subExecSqlContext.setExecType(SqlConstantUtils.SQL_EXEC_DDL);
TransmittableThreadLocal<ExecSqlContext> ttl = new TransmittableThreadLocal<>();
ttl.set(subExecSqlContext);
executorService.execute(new SubThreadTask(ttl, retMsg, downLatch));
}
}
try {
......@@ -999,6 +1015,7 @@ public class SysDpExecuteHistoryServiceImpl extends BaseServiceImpl<SysDpExecute
} catch (InterruptedException ie) {
logger.info("==> 错误信息：[{}]", ie.getMessage());
}
// subExecContext.getMsg().forEach((k,v) -> result.add("第" + k + "行语句，" + v));
retMsg.forEach((k, v) -> result.add("第" + k + "行语句，" + v));
if (!result.isEmpty()) {
responseData.setRows(result);
......@@ -1008,12 +1025,12 @@ public class SysDpExecuteHistoryServiceImpl extends BaseServiceImpl<SysDpExecute
}
private static class SubThreadTask implements Runnable {
private final ReentrantLock mainLock = new ReentrantLock();
private final SubExecSqlContext subExecContext;
private final TransmittableThreadLocal<ExecSqlContext> ttl;
private final ConcurrentHashMap<Integer, String> retMsg;
private final CountDownLatch downLatch;
private SubThreadTask(SubExecSqlContext subExecContext, ConcurrentHashMap<Integer, String> retMsg, CountDownLatch downLatch) {
this.subExecContext = subExecContext;
private SubThreadTask(TransmittableThreadLocal<ExecSqlContext> ttl, ConcurrentHashMap<Integer, String> retMsg, CountDownLatch downLatch) {
this.ttl = ttl;
this.retMsg = retMsg;
this.downLatch = downLatch;
}
......@@ -1022,13 +1039,13 @@ public class SysDpExecuteHistoryServiceImpl extends BaseServiceImpl<SysDpExecute
public void run() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
final Integer seq = subExecContext.getSeq();
final String sql = subExecContext.getSql();
final Integer seq = ttl.get().getSeq();
final String sql = ttl.get().getSql();
try {
final SysDpExecuteHistoryService service = subExecContext.getService();
final IRequest requestContext = subExecContext.getRequestContext();
final String ip = subExecContext.getIp();
final String execType = subExecContext.getExecType();
final SysDpExecuteHistoryService service = ttl.get().getService();
final IRequest requestContext = ttl.get().getRequestContext();
final String ip = ttl.get().getIp();
final String execType = ttl.get().getExecType();
ResponseData subResponseData = new ResponseData();
if (StringUtils.equals(execType, SqlConstantUtils.SQL_EXEC_INSERT)) {
subResponseData = service.insert(requestContext, ip, sql);
......
v2-多线程版本/java-code/dp/util/IPUtils.java
View file @ fb93c651
......@@ -3,9 +3,12 @@ package com.hand.hls.dp.util;
import org.apache.commons.lang3.StringUtils;
import javax.servlet.http.HttpServletRequest;
import java.math.BigInteger;
import java.net.InetAddress;
import java.nio.Buffer;
import java.util.Arrays;
import java.util.Objects;
import java.util.stream.Collectors;
/**
* @author <a href="mailto:zhangnet14@gmail.com">Spencer Chang</a>
......@@ -15,8 +18,8 @@ import java.util.Objects;
public class IPUtils {
private static final String IPV4_LOCAL_HOST = "127.0.0.1";
private static final String IPV6_LOCAL_HOST = "0:0:0:0:0:0:0:1";
public static String getIpAddress(HttpServletRequest request) {
boolean ipv6 = false;
if (Objects.isNull(request)) {
return "";
}
......@@ -43,41 +46,40 @@ public class IPUtils {
if (Objects.isNull(ip) || ip.length() == 0 || "unknown".equalsIgnoreCase(ip)) {
ip = request.getRemoteAddr();
// 本地局域网放行
if (Objects.equals(IPV4_LOCAL_HOST,ip) || Objects.equals(IPV6_LOCAL_HOST,ip)) {
if (Objects.equals(IPV4_LOCAL_HOST, ip) || Objects.equals(IPV6_LOCAL_HOST, ip)) {
// 根据网卡获取本机配置的ip
InetAddress inetAddress = null;
try {
inetAddress = InetAddress.getLocalHost();
ip = inetAddress.getHostAddress();
} catch (Exception e) {
// 默认
ip = IPV4_LOCAL_HOST;
}
byte[] address = inetAddress.getAddress();
if (address.length == 16) {
// IPV6
ip = inetAddress.getHostAddress();
ipv6 = true;
} else if (address.length == 4) {
// IPV4
ip = inetAddress.getHostAddress();
} else {
//throw new Exception("Invalid IPv6 address: '" + ip + '\'');
// 默认
ip = IPV6_LOCAL_HOST;
ipv6 = true;
}
}
}
// 对于通过多个代理的情况，第一个IP为客户端真实IP,多个IP按照','分割
// "***.***.***.***".length() = 15
if (Objects.nonNull(ip) && ip.length() > 15 && !ipv6) {
if (Objects.nonNull(ip) && ip.length() > 15) {
if (ip.indexOf(",") > 0) {
ip = ip.substring(0, ip.indexOf(","));
}
}
return extractString(ip);
}
// 获取本机ip
public static String getIpAddressLocal() {
String ip = IPV4_LOCAL_HOST;
// 根据网卡获取本机配置的ip
InetAddress inetAddress = null;
try {
inetAddress = InetAddress.getLocalHost();
ip = inetAddress.getHostAddress();
} catch (Exception e) {
// ignore
}
return ip;
}
/**
* 获取访问主机的网段（4位）
......
v2-多线程版本/java-code/dp/util/LogUtils.java
View file @ fb93c651
......@@ -74,6 +74,7 @@ public class LogUtils {
buffer.get(bytes);
// 内容写入ZipOutputStream
zos.write(bytes,0,n);
zos.flush();
}
// 切换buffer写模式
buffer.clear();
......@@ -81,12 +82,12 @@ public class LogUtils {
} catch (FileNotFoundException fe){
// 文件没找到
if (logger.isErrorEnabled()) {
logger.error("{} file not found！", filePath);
logger.error("[{}] file not found！", filePath);
}
} catch (IOException e){
// io例外
if (logger.isWarnEnabled()) {
logger.warn("{} file IOException！Message:[{}]", filePath, e.getMessage());
logger.warn("[{}] file IOException！Message:[{}]", filePath, e.getMessage());
}
}
}
......
v2-多线程版本/java-code/dp/util/Maps.java 0 → 100644
View file @ fb93c651
package com.hand.hls.dp.util;
import java.util.HashMap;
/**
* @author <a href="mailto:zhangnet14@gmail.com">Spencer Chang</a>
* @date 2020/9/18 16:11
* @see com.google.common.collect.Maps#newHashMapWithExpectedSize
*/
public final class Maps {
/**
* The maximum capacity, used if a higher value is implicitly specified
* by either of the constructors with arguments.
* MUST be a power of two <= 1<<30.
*/
static final int MAXIMUM_CAPACITY = 1 << 30;
/**
* The load factor used when none specified in constructor.
*/
static final float DEFAULT_LOAD_FACTOR = 0.75f;
/**
* default expected size
*/
static final int DEFAULT_EXPECTED_SIZE = 3;
/**
* Creates a {@code HashMap} instance, with a high enough "initial capacity" that it <i>should</i>
* hold {@code expectedSize} elements without growth. This behavior cannot be broadly guaranteed,
* but it is observed to be true for OpenJDK 1.7. It also can't be guaranteed that the method
* isn't inadvertently <i>oversizing</i> the returned map.
*
* @param expectedSize the number of entries you expect to add to the returned map
* @return a new, empty {@code HashMap} with enough capacity to hold {@code expectedSize} entries
* without resizing
* @throws IllegalArgumentException if {@code expectedSize} is negative
*/
public static <K, V> HashMap<K, V> newHashMapWithExpectedSize(int expectedSize) {
return new HashMap<>(capacity(expectedSize));
}
static int capacity(int expectedSize) {
if (expectedSize < DEFAULT_EXPECTED_SIZE) {
if (expectedSize < 0) {
throw new IllegalArgumentException("expectedSize cannot be negative but was: " + expectedSize);
}
return expectedSize + 1;
} else {
return expectedSize < MAXIMUM_CAPACITY ? (int) ((float) expectedSize / DEFAULT_LOAD_FACTOR + 1.0f) : Integer.MAX_VALUE;
}
}
}
\ No newline at end of file
v2-多线程版本/lview-code/DP/DP001/sys_dp.lview
View file @ fb93c651
......@@ -18,7 +18,7 @@
form.method = "post";
let url = '${/request/@context_path}/sys/dp/query?ck=1';
let token = $jq('meta[name=_csrf]').attr('content');
form.action = url + (url.indexOf('?') == -1 ? '?' : '&') + 'r=' + Math.random() + '&_csrf=' + token;
form.action = url + (url.indexOf('?') === -1 ? '?' : '&') + 'r=' + Math.random() + '&_csrf=' + token;
let iframe = Ext.get('_export_dp_query_window') || new Ext.Template('<iframe id ="_export_dp_query_window" name="_export_dp_query_window" style="position:absolute;left:-1000px;top:-1000px;width:1px;height:1px;display:none"></iframe>').insertFirst(document.body, {}, true)
var s = document.createElement("input");
s.id = "_request_data";
......@@ -112,38 +112,12 @@
Leaf.Masker.unmask(Ext.getBody());
}, 200, 100);
}
function download_log() {
$('dp001_download_log_btn').disable();
Leaf.showConfirm('${l:HLS.PROMPT}', '确认下载日志吗？', function okFun() {
Leaf.Masker.mask(Ext.getBody(), '正在下载中...');
let form = document.createElement("form");
form.target = "_export_dp_log_window";
form.method = "post";
let url = '${/request/@context_path}/sys/dp/download/log';
let token = $jq('meta[name=_csrf]').attr('content');
form.action = url + (url.indexOf('?') == -1 ? '?' : '&') + 'r=' + Math.random() + '&_csrf=' + token;
let iframe = Ext.get('_export_dp_log_window') || new Ext.Template('<iframe id ="_export_dp_log_window" name="_export_dp_log_window" style="position:absolute;left:-1000px;top:-1000px;width:1px;height:1px;display:none"></iframe>').insertFirst(document.body, {}, true)
document.body.appendChild(form);
form.submit();
Ext.fly(form).remove();
setTimeout(function () {
$('dp001_download_log_btn').enable();
Leaf.Masker.unmask(Ext.getBody());
}, 10000);
}, function cancel() {
$('dp001_download_log_btn').enable();
Leaf.Masker.unmask(Ext.getBody());
}, 200, 100);
}
]]></script>
<a:screenBody>
<a:screenTopToolbar>
<a:gridButton id="dp001_query_btn" click="dp001_query" text="查询"/>
<a:gridButton id="dp001_exec_btn" click="dp001_exec" text="执行"/>
<a:gridButton id="dp001_download_cluster_log_btn" click="download_cluster_log" text="日志下载(当天集群环境日志)"/>
<a:gridButton id="dp001_download_log_btn" click="download_log" text="日志下载(7天当前请求环境日志)"/>
</a:screenTopToolbar>
<a:fieldSet title="待执行">
<a:textArea id="context_tta" name="context" marginWidth="50" marginHeight="450"/>
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment