kris/data-dump
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

【feat】 优化ORG连接问题

Browse Source
This commit is contained in:
Kris 2025-08-19 10:44:47 +08:00
parent 24474a2c8a
commit b4607439a3
2 changed files with 151 additions and 128 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

134
src/main/java/com/celnet/datadump/config/SalesforceConnect.java
View File

@ -18,6 +18,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.TimeUnit;
/**
* @author Red
@ -31,45 +32,57 @@ public class SalesforceConnect {
@Resource
private CustomMapper customerMapper;
/**
* 创建连接
* @return
*/
public PartnerConnection createConnect() {
try {
List<Map<String, Object>> poll = customerMapper.list("code,value","org_config",null);
//遍历poll,找出code值为SOURCE_ORG_URL，SOURCE_ORG_USERNAME，SOURCE_ORG_PASSWORD的value值
Map<String, String> map = new HashMap<>();
for (Map<String, Object> map1 : poll) {
if ("SOURCE_ORG_URL".equals(map1.get("code"))) {
map.put("url", (String) map1.get("value"));
public PartnerConnection createConnect() {
int failCount = 0;
//重试连接次数
while (true){
try {
List<Map<String, Object>> poll = customerMapper.list("code,value","org_config",null);
//遍历poll,找出code值为SOURCE_ORG_URL，SOURCE_ORG_USERNAME，SOURCE_ORG_PASSWORD的value值
Map<String, String> map = new HashMap<>();
for (Map<String, Object> map1 : poll) {
if ("SOURCE_ORG_URL".equals(map1.get("code"))) {
map.put("url", (String) map1.get("value"));
}
if ("SOURCE_ORG_USERNAME".equals(map1.get("code"))) {
map.put("username", (String) map1.get("value"));
}
if ("SOURCE_ORG_PASSWORD".equals(map1.get("code"))) {
map.put("password", String.valueOf(map1.get("value")));
}
}
if ("SOURCE_ORG_USERNAME".equals(map1.get("code"))) {
map.put("username", (String) map1.get("value"));
String username = map.get("username");
ConnectorConfig config = new ConnectorConfig();
config.setUsername(username);
config.setPassword(map.get("password"));
String url = map.get("url");
config.setAuthEndpoint(url);
config.setServiceEndpoint(url);
config.setConnectionTimeout(60 * 60 * 1000);
config.setReadTimeout(60 * 60 * 1000);
return new PartnerConnection(config);
} catch (ConnectionException e) {
failCount ++;
log.error("源ORG连接异常！休眠一分钟再次发起重试~~", e);
try {
TimeUnit.MINUTES.sleep(1);
}catch (InterruptedException ie) {
log.error("休眠异常", ie);
}
if ("SOURCE_ORG_PASSWORD".equals(map1.get("code"))) {
map.put("password", String.valueOf(map1.get("value")));
if (failCount > 3) {
String message = "源ORG三次重试连接错误！";
String format = String.format("ORG连接异常！, \ncause:\n%s", message);
EmailUtil.send("DataDump ERROR", format);
log.error("exception message", e);
}
}
String username = map.get("username");
ConnectorConfig config = new ConnectorConfig();
config.setUsername(username);
config.setPassword(map.get("password"));
String url = map.get("url");
config.setAuthEndpoint(url);
config.setServiceEndpoint(url);
config.setConnectionTimeout(60 * 60 * 1000);
config.setReadTimeout(60 * 60 * 1000);
return new PartnerConnection(config);
} catch (ConnectionException e) {
String message = "源ORG连接配置错误！";
String format = String.format("ORG连接异常！, \ncause:\n%s", message);
EmailUtil.send("DataDump ERROR", format);
log.error("exception message", e);
return null;
}
}
/**
@ -77,41 +90,38 @@ public class SalesforceConnect {
* @author kris
*/
public BulkConnection createBulkConnect() {
try {
List<Map<String, Object>> poll = customerMapper.list("code,value","org_config",null);
//遍历poll,找出code值为SOURCE_ORG_URL，SOURCE_ORG_USERNAME，SOURCE_ORG_PASSWORD的value值
Map<String, String> map = new HashMap<>();
for (Map<String, Object> map1 : poll) {
if ("SOURCE_ORG_URL".equals(map1.get("code"))) {
map.put("url", (String) map1.get("value"));
int failCount = 0;
while (true){
try {
List<Map<String, Object>> poll = customerMapper.list("code,value","org_config",null);
//遍历poll,找出code值为SOURCE_ORG_URL，SOURCE_ORG_USERNAME，SOURCE_ORG_PASSWORD的value值
Map<String, String> map = new HashMap<>();
for (Map<String, Object> map1 : poll) {
if ("SOURCE_ORG_URL".equals(map1.get("code"))) {
map.put("url", (String) map1.get("value"));
}
if ("SOURCE_ORG_USERNAME".equals(map1.get("code"))) {
map.put("username", (String) map1.get("value"));
}
if ("SOURCE_ORG_PASSWORD".equals(map1.get("code"))) {
map.put("password", String.valueOf(map1.get("value")));
}
}
if ("SOURCE_ORG_USERNAME".equals(map1.get("code"))) {
map.put("username", (String) map1.get("value"));
}
if ("SOURCE_ORG_PASSWORD".equals(map1.get("code"))) {
map.put("password", String.valueOf(map1.get("value")));
return BulkUtil.getBulkConnection(map.get("username"),map.get("password"),map.get("url"));
} catch (Exception e) {
failCount ++;
log.error("源ORG连接异常！休眠一分钟再次发起重试~~", e);
try {
TimeUnit.MINUTES.sleep(1);
}catch (InterruptedException ie) {
log.error("休眠异常", ie);
} if (failCount > 3) {
String message = "源ORG三次重试连接错误！";
String format = String.format("ORG连接异常！, \ncause:\n%s", message);
EmailUtil.send("DataDump ERROR", format);
log.error("exception message", e);
}
}
//
// String username = map.get("username");
// ConnectorConfig config = new ConnectorConfig();
// config.setUsername(username);
// config.setPassword(map.get("password"));
// String url = map.get("url");
// config.setAuthEndpoint(url);
// config.setServiceEndpoint(url);
// config.setConnectionTimeout(60 * 60 * 1000);
// config.setReadTimeout(60 * 60 * 1000);
// PartnerConnection connection = new PartnerConnection(config);
// config.setRestEndpoint(url);
// config.setSessionId(connection.getSessionHeader().getSessionId());
return BulkUtil.getBulkConnection(map.get("username"),map.get("password"),map.get("url"));
} catch (Exception e) {
String message = "源ORG连接配置错误！";
String format = String.format("ORG连接异常！, \ncause:\n%s", message);
EmailUtil.send("DataDump ERROR", format);
log.error("exception message", e);
return null;
}
}

145
src/main/java/com/celnet/datadump/config/SalesforceTargetConnect.java
View File

@ -18,6 +18,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.TimeUnit;
@Component
@Slf4j
@ -30,42 +31,56 @@ public class SalesforceTargetConnect {
* 创建连接
* @return
*/
public PartnerConnection createConnect() {
try {
List<Map<String, Object>> poll = customerMapper.list("code,value","org_config",null);
//遍历poll,找出code值为TARGET_ORG_URL，TARGET_ORG_USERNAME，TARGET_ORG_PASSWORD的value值
Map<String, String> map = new HashMap<>();
for (Map<String, Object> map1 : poll) {
if ("TARGET_ORG_URL".equals(map1.get("code"))) {
map.put("url", (String) map1.get("value"));
public PartnerConnection createConnect() {
int failCount = 0;
//重试连接次数
while (true){
try {
List<Map<String, Object>> poll = customerMapper.list("code,value","org_config",null);
//遍历poll,找出code值为TARGET_ORG_URL，TARGET_ORG_USERNAME，TARGET_ORG_PASSWORD的value值
Map<String, String> map = new HashMap<>();
for (Map<String, Object> map1 : poll) {
if ("TARGET_ORG_URL".equals(map1.get("code"))) {
map.put("url", (String) map1.get("value"));
}
if ("TARGET_ORG_USERNAME".equals(map1.get("code"))) {
map.put("username", (String) map1.get("value"));
}
if ("TARGET_ORG_PASSWORD".equals(map1.get("code"))) {
map.put("password", (String) map1.get("value"));
}
}
if ("TARGET_ORG_USERNAME".equals(map1.get("code"))) {
map.put("username", (String) map1.get("value"));
}
if ("TARGET_ORG_PASSWORD".equals(map1.get("code"))) {
map.put("password", (String) map1.get("value"));
}
}
String username = map.get("username");
ConnectorConfig config = new ConnectorConfig();
config.setUsername(username);
config.setPassword(map.get("password"));
String url = map.get("url");
config.setAuthEndpoint(url);
config.setServiceEndpoint(url);
config.setConnectionTimeout(60 * 60 * 1000);
config.setReadTimeout(60 * 60 * 1000);
PartnerConnection connection = new PartnerConnection(config);
String orgId = connection.getUserInfo().getOrganizationId();
return connection;
} catch (ConnectionException e) {
String message = "目标ORG连接配置错误！";
String format = String.format("ORG连接异常！, \ncause:\n%s", message);
EmailUtil.send("DataDump ERROR", format);
log.error("exception message", e);
return null;
String username = map.get("username");
ConnectorConfig config = new ConnectorConfig();
config.setUsername(username);
config.setPassword(map.get("password"));
String url = map.get("url");
config.setAuthEndpoint(url);
config.setServiceEndpoint(url);
config.setConnectionTimeout(60 * 60 * 1000);
config.setReadTimeout(60 * 60 * 1000);
PartnerConnection connection = new PartnerConnection(config);
String orgId = connection.getUserInfo().getOrganizationId();
return connection;
} catch (ConnectionException e) {
failCount++;
log.error("目标ORG连接异常！休眠一分钟再次发起重试~~", e);
try {
TimeUnit.MINUTES.sleep(1);
}catch (InterruptedException ie) {
log.error("休眠异常", ie);
} if (failCount > 3) {
String message = "目标ORG三次重试连接错误！";
String format = String.format("ORG连接异常！, \ncause:\n%s", message);
EmailUtil.send("DataDump ERROR", format);
log.error("exception message", e);
return null;
}
}
}
}
/**
@ -73,42 +88,40 @@ public class SalesforceTargetConnect {
* @author kris
*/
public BulkConnection createBulkConnect() {
try {
List<Map<String, Object>> poll = customerMapper.list("code,value","org_config",null);
//遍历poll,找出code值为TARGET_ORG_URL，TARGET_ORG_USERNAME，TARGET_ORG_PASSWORD的value值
Map<String, String> map = new HashMap<>();
for (Map<String, Object> map1 : poll) {
if ("TARGET_ORG_URL".equals(map1.get("code"))) {
map.put("url", (String) map1.get("value"));
int failCount = 0;
while (true){
//重试连接次数
try {
List<Map<String, Object>> poll = customerMapper.list("code,value","org_config",null);
//遍历poll,找出code值为TARGET_ORG_URL，TARGET_ORG_USERNAME，TARGET_ORG_PASSWORD的value值
Map<String, String> map = new HashMap<>();
for (Map<String, Object> map1 : poll) {
if ("TARGET_ORG_URL".equals(map1.get("code"))) {
map.put("url", (String) map1.get("value"));
}
if ("TARGET_ORG_USERNAME".equals(map1.get("code"))) {
map.put("username", (String) map1.get("value"));
}
if ("TARGET_ORG_PASSWORD".equals(map1.get("code"))) {
map.put("password", (String) map1.get("value"));
}
}
if ("TARGET_ORG_USERNAME".equals(map1.get("code"))) {
map.put("username", (String) map1.get("value"));
}
if ("TARGET_ORG_PASSWORD".equals(map1.get("code"))) {
map.put("password", (String) map1.get("value"));
return BulkUtil.getBulkConnection(map.get("username"),map.get("password"),map.get("url"));
} catch (Exception e) {
failCount++;
log.error("目标ORG连接异常！休眠一分钟再次发起重试~~", e);
try {
TimeUnit.MINUTES.sleep(1);
}catch (InterruptedException ie) {
log.error("休眠异常", ie);
} if (failCount > 3) {
String message = "目标ORG三次重试连接错误！";
String format = String.format("ORG连接异常！, \ncause:\n%s", message);
EmailUtil.send("DataDump ERROR", format);
log.error("exception message", e);
return null;
}
}
// String username = ;
// ConnectorConfig config = new ConnectorConfig();
// config.setUsername(username);
// config.setPassword();
// String url = ;
// config.setAuthEndpoint(url);
// config.setServiceEndpoint(url);
// config.setConnectionTimeout(60 * 60 * 1000);
// config.setReadTimeout(60 * 60 * 1000);
// PartnerConnection connection = new PartnerConnection(config);
// config.setRestEndpoint(url);
// config.setSessionId(connection.getSessionHeader().getSessionId());
return BulkUtil.getBulkConnection(map.get("username"),map.get("password"),map.get("url"));
} catch (Exception e) {
String message = "目标ORG连接配置错误！";
String format = String.format("ORG连接异常！, \ncause:\n%s", message);
EmailUtil.send("DataDump ERROR", format);
log.error("exception message", e);
return null;
}
}