Sunyard/sydapi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix:主备切换测试

Browse Source
This commit is contained in:
waner
2025-05-30 16:21:39 +08:00
parent 085d71feb9
commit fbcbe79388
11 changed files with 543 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/Sample/TestCert.java
-76
View File
@@ -1,76 +0,0 @@
package com.Sample;
import cn.hutool.core.date.StopWatch;
import com.google.gson.Gson;
import racal.sunyard.main.SydApi4j;
import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.*;
public class TestCert {
public static void main(String[] args) throws Exception {
StopWatch stopWatch = new StopWatch();
stopWatch.start();
System.out.println("-----------start---------------");
SydApi4j sydApi4j = new SydApi4j();
String ipInfo = "/app/test/ipInfo.txt";
String snmpInfo = "/app/test/snmp.txt";
String certInfo = "/app/test/cert.txt";
Map<String, String> map = readFileToMap(ipInfo);
String ipInfos = map.getOrDefault("ipList", null);
List<String> ipList = new ArrayList<>();
if (ipInfos != null) {
ipList = Arrays.asList(ipInfos.split(","));
}
int timeout = Integer.parseInt(map.get("timout"));
String cert = readFileAsString(certInfo);
String bkCode = map.get("bkcode");
Map<String, String> snmpMap = readFileToMap(snmpInfo);
Gson gson = new Gson();
sydApi4j.importCertToServers(ipList, timeout, cert, bkCode, gson.toJson(snmpMap));
System.out.println("------------end-----------------------");
stopWatch.stop();
System.out.println("耗时:" + stopWatch.getTotalTimeMillis() + "ms");
}
public static Map<String, String> readFileToMap(String filePath) {
Map<String, String> map = new HashMap<>();
try (BufferedReader br = new BufferedReader(new FileReader(filePath))) {
String line;
while ((line = br.readLine()) != null) {
// 跳过空行和注释行（假设以#开头的为注释）
if (line.trim().isEmpty() || line.startsWith("#")) {
continue;
}
// 按分隔符分割键值（例如"="或":"）
String[] parts = line.split("=", 2); // 使用第二个参数限制分割次数
if (parts.length == 2) {
String key = parts[0].trim();
String value = parts[1].trim();
map.put(key, value);
}
}
} catch (IOException e) {
e.printStackTrace();
}
return map;
}
public static String readFileAsString(String filePath) {
StringBuilder content = new StringBuilder();
try (BufferedReader br = new BufferedReader(new FileReader(filePath))) {
String line;
while ((line = br.readLine()) != null) {
content.append(line).append("\n");
}
} catch (IOException e) {
e.printStackTrace();
}
return content.toString();
}
}
src/main/java/com/jmeter/test/Test4Jemter.java
+67 -1
View File
@@ -1,5 +1,6 @@
package com.jmeter.test;
import com.google.gson.Gson;
import com.sunyard.SydApiException;
import com.sunyard.proto.Util;
import org.apache.jmeter.config.Arguments;
@@ -10,6 +11,10 @@ import racal.sunyard.main.SydApi4j;
import racal.sunyard.main.SydApiBuilder;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
public class Test4Jemter extends AbstractJavaSamplerClient {
@@ -30,6 +35,8 @@ public class Test4Jemter extends AbstractJavaSamplerClient {
private static ThreadLocal<SydApi4j> api;
private int dataLength;
/**
* 定义默认的参数，在jmeter中会自动显示在这里定义的默认参数
@@ -44,6 +51,7 @@ public class Test4Jemter extends AbstractJavaSamplerClient {
arguments.addArgument("ip", "172.1.41.139");
// arguments.addArgument("port", "8891");
// arguments.addArgument("ip2", "172.16.17.161");
arguments.addArgument("dataLength", "1024");
arguments.addArgument("type", "1");
arguments.addArgument("sign", "12345678");
arguments.addArgument("detachedSign", "12345678");
@@ -63,6 +71,7 @@ public class Test4Jemter extends AbstractJavaSamplerClient {
ip = context.getParameter("ip");
// port = Integer.parseInt(context.getParameter("8891"));
// ip2 = context.getParameter("ip2");
dataLength = context.getIntParameter("dataLength");
type = context.getIntParameter("type");
sign = context.getParameter("sign");
detachedSign = context.getParameter("detachedSign");
@@ -92,7 +101,8 @@ public class Test4Jemter extends AbstractJavaSamplerClient {
//设置开始标志，否则无法统计tps等
sr.sampleStart();
try {
orgData = oData.getBytes(); //SM2裸签
// orgData = oData.getBytes(); //SM2裸签
orgData = new byte[dataLength];
//执行业务逻辑
switch(type){
case 1:
@@ -205,6 +215,14 @@ public class Test4Jemter extends AbstractJavaSamplerClient {
case 17:
sr.setResponseCodeOK();
break;
case 18:
testAhnx();
sr.setResponseCodeOK();
break;
case 19:
testImportCert();
sr.setResponseCodeOK();
break;
default:
sr.setResponseCodeOK();
sr.setResponseData("请选择测试组", StandardCharsets.UTF_8.displayName());
@@ -222,6 +240,54 @@ public class Test4Jemter extends AbstractJavaSamplerClient {
return sr;
}
public void testAhnx() {
orgData = new byte[1024];
//裸签
byte[] nakedSign = api.get().SYD_NakedSign(orgData,SM2dn);
//裸签验签
boolean ver = api.get().SYD_NakedVerify(orgData,nakedSign, SM2dn);
//detach签名
String detachedSign = api.get().detachedSign(orgData, SM2dn);
//detach验签
boolean ver1 = api.get().detachedVerify(orgData, detachedSign);
testImportCert();
}
public void testImportCert() {
String cert = "-----BEGIN CERTIFICATE-----\n" +
"MIICMDCCAdagAwIBAgIJAOWoGwJCnjlLMAoGCCqBHM9VAYN1MGcxCzAJBgNVBAYT\n" +
"AkNOMRAwDgYDVQQIDAdCZWlqaW5nMRAwDgYDVQQHDAdIYWlEaWFuMRMwEQYDVQQK\n" +
"DApHTUNlcnQub3JnMR8wHQYDVQQDDBZHTUNlcnQgR00gUm9vdCBDQSAtIDAxMB4X\n" +
"DTI1MDUxMzAyMzUxM1oXDTI2MDUxMzAyMzUxM1owRTELMAkGA1UEBgwCQ04xCzAJ\n" +
"BgNVBAgMAnp6MQswCQYDVQQHDAJ6ejEMMAoGA1UECgwDenp6MQ4wDAYDVQQDDAV6\n" +
"enp6ejBZMBMGByqGSM49AgEGCCqBHM9VAYItA0IABK5fyRSACWPIBBr/3U+qvjvp\n" +
"niropv2PrA3ZLU1bG8vQpwAEMA0kwyqGgXURBPbvWdAhN7s9wfvHbJyMd9CMZGqj\n" +
"gYwwgYkwDAYDVR0TAQH/BAIwADALBgNVHQ8EBAMCB4AwLAYJYIZIAYb4QgENBB8W\n" +
"HUdNQ2VydC5vcmcgU2lnbmVkIENlcnRpZmljYXRlMB0GA1UdDgQWBBSQB7UUqEuG\n" +
"kl4mrx3MAhKTptfbYjAfBgNVHSMEGDAWgBR/Wl47AIRZKg+YvqEObzmVQxBNBzAK\n" +
"BggqgRzPVQGDdQNIADBFAiEA4kgTdn/iA/XYfjTEeOcELwBXd0tTFghkpfP8DhBF\n" +
"KFQCIAXZTL6VPtW67tM/DF4ZgFt5S/YIiFombKa2bWOfL81u\n" +
"-----END CERTIFICATE-----";
Map<String, String> map = new HashMap<>();
map.put("snmp.target.ip", "172.1.41.191");
map.put("snmp.target.port", "162");
map.put("enterprise.oid", "1.3.6.1.4.1.9999");
map.put("snmp.community", "public");
map.put("snmp.version", "1");
List<String> masterNodes = new ArrayList<>();
masterNodes.add("172.1.41.139:8889");
String snmpConfig = new Gson().toJson(map);
SydApi4j sydApi4j = new SydApi4j();
sydApi4j.importCertToServers(masterNodes, 100, cert, "12345678", snmpConfig);
}
/**
* 测试完成后执行，一般用于关闭资源
* @param context 上下文对象，可用于获取jmeter传递的参数
src/main/java/com/sunyard/pool/ConnectionManager.java
+135
View File
@@ -0,0 +1,135 @@
package com.sunyard.pool;
import racal.sunyard.main.SydApi4j;
import java.io.IOException;
import java.util.Set;
import java.util.concurrent.*;
public class ConnectionManager {
// 单例实例（静态内部类实现线程安全延迟加载）
private static class Holder {
static final ConnectionManager INSTANCE = new ConnectionManager();
}
// 私有构造函数（禁止外部实例化）
private ConnectionManager() {
Runtime.getRuntime().addShutdownHook(new Thread(this::shutdown));
}
// 全局访问点
public static ConnectionManager getInstance() {
return Holder.INSTANCE;
}
// 使用线程安全的ConcurrentHashMap存储连接
private final ConcurrentHashMap<String, SydApi4j> connectionCache = new ConcurrentHashMap<>();
/**
* 获取或创建指定URL的连接
* @param url 目标服务地址
* @return 已存在的或新建的SydReq连接
* @throws IOException 当创建连接失败时抛出
*/
public SydApi4j getConnection(String url, int timeout) throws Exception {
// 使用compute方法保证原子操作
return connectionCache.compute(url, (key, existingConn) -> {
try {
// 验证现有连接有效性
if (isConnectionValid(existingConn)) {
return existingConn;
}
// 关闭失效连接（如果存在）
closeConnectionQuietly(existingConn);
// 创建新连接
return createNewConnection(url, timeout);
} catch (Exception e) {
throw new RuntimeException("连接创建失败", e);
}
});
}
// 连接有效性验证
private boolean isConnectionValid(SydApi4j conn) {
return conn != null && conn.isActive();
}
// 安全关闭连接
private void closeConnectionQuietly(SydApi4j conn) {
try {
if (conn != null) {
conn.disconnect();
}
} catch (Exception e) {
}
}
// 创建新连接（提取共用逻辑）
private SydApi4j createNewConnection(String url, int timeout) throws Exception {
try {
String[] addressParts = parseUrl(url);
String ip = addressParts[0];
int port = Integer.parseInt(addressParts[1]);
SydApi4j newConn = new SydApi4j();
newConn.connect(ip, port, null, timeout);
return newConn;
} catch (Exception e) {
throw new RuntimeException("连接创建失败: " + url, e);
}
}
// URL解析验证
private String[] parseUrl(String url) throws IllegalArgumentException {
String[] parts = url.split(":");
if (parts.length != 2) {
throw new IllegalArgumentException("无效的URL格式: " + url);
}
return parts;
}
/**
* 关闭指定URL的连接并从缓存移除
* @param url 需要关闭的服务地址
*/
public void closeConnection(String url) {
SydApi4j conn = connectionCache.remove(url);
if (conn != null) {
try {
conn.disconnect(); // 假设这是关闭连接的方法
System.out.println("连接已关闭: " + url);
} catch (Exception e) {
System.err.println("关闭连接异常 [" + url + "]: " + e.getMessage());
}
}
}
/**
* 关闭所有连接并清空缓存
*/
public void closeAllConnections() {
// 创建当前URL的副本避免并发修改
Set<String> urls = ConcurrentHashMap.newKeySet();
urls.addAll(connectionCache.keySet());
urls.forEach(this::closeConnection);
}
/**
* 获取当前缓存中的连接数量
*/
public int size() {
return connectionCache.size();
}
private void shutdown() {
connectionCache.forEach((url, conn) -> {
conn.disconnect();
});
connectionCache.clear();
}
}
src/main/java/com/sunyard/snmp/SnmpTrapSender.java
+8 -1
View File
@@ -9,6 +9,7 @@ import org.snmp4j.smi.*;
import org.snmp4j.transport.DefaultUdpTransportMapping;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.Properties;
import java.util.Vector;
@@ -79,8 +80,14 @@ public class SnmpTrapSender implements AutoCloseable{
new TimeTicks(System.currentTimeMillis()/1000)));
pdu.add(new VariableBinding(SnmpConstants.snmpTrapOID,
new OID(enterpriseOid + ".1.1")));
byte[] msg = new byte[10];
try {
msg = message.getBytes("GBK");
} catch (Exception e) {
System.out.println("编码错误");
}
pdu.add(new VariableBinding(new OID(enterpriseOid + ".2.1"),
new OctetString(message)));
new OctetString(msg)));
pdu.add(new VariableBinding(new OID(enterpriseOid + ".2.2"),
new Integer32(severity)));
}
src/main/java/com/sunyard/task/AsyncTaskExecutor.java
+7 -28
View File
@@ -1,11 +1,7 @@
package com.sunyard.task;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.function.Supplier;
import java.util.stream.Collectors;
public class AsyncTaskExecutor {
@@ -20,24 +16,7 @@ public class AsyncTaskExecutor {
// 默认配置构造函数
public AsyncTaskExecutor() {
this(4, 2);
}
public List<TaskOutcome<String>> execTasks(List<TaskContext<String>> taskContexts) {
List<CompletableFuture<String>> futures =
taskContexts.stream().map(this::retryWithDelay).collect(Collectors.toList());
List<TaskOutcome<String>> taskOutcomes = new ArrayList<>();
taskContexts.forEach(task ->
retryWithDelay(task).whenCompleteAsync((result, ex) -> {
if (ex != null) {
// 提取原始异常
Throwable rootCause = ex instanceof CompletionException ? ex.getCause() : ex;
taskOutcomes.add(new TaskOutcome<>(task, rootCause));
} else {
taskOutcomes.add(new TaskOutcome<>(task, result));
}
}));
return taskOutcomes;
this(2, 2);
}
public List<TaskOutcome<String>> execBatchTasks(List<TaskContext<String>> taskContexts) {
@@ -94,12 +73,12 @@ public class AsyncTaskExecutor {
private <T> CompletableFuture<T> handleRetry(TaskContext<T> context, Throwable ex) {
if (context.canRetry()) {
context.decreaseRetry();
System.out.printf("[重试调度] ID: %s| IP:%s | 延迟: %ss | 剩余重试: %d | 错误: %s%n",
context.getTaskId(),
context.getAddress(),
context.getRetryDelay().getSeconds(),
context.getRetriesLeft(),
ex.getCause().getMessage());
// System.out.printf("[重试调度] ID: %s| IP:%s | 延迟: %ss | 剩余重试: %d | 错误: %s%n",
// context.getTaskId(),
// context.getAddress(),
// context.getRetryDelay().getSeconds(),
// context.getRetriesLeft(),
// ex.getCause().getMessage());
// 创建延迟重试的Future
CompletableFuture<T> delayedRetry = new CompletableFuture<>();
src/main/java/com/sunyard/util/IpLockUtils.java
+107
View File
@@ -0,0 +1,107 @@
package com.sunyard.util;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
/**
* IP锁工具类 - 高并发安全实现
*
* 为每个IP地址提供唯一的可重入锁，确保同一IP的操作顺序执行
*/
public class IpLockUtils {
private static final ConcurrentMap<String, ReentrantLock> LOCK_MAP = new ConcurrentHashMap<>();
private static final Object CREATION_LOCK = new Object();
/**
* 获取IP对应的锁（保证唯一性）
* @param ip IP地址
* @return 可重入锁实例
*/
public static ReentrantLock getLockForIp(String ip) {
ReentrantLock lock = LOCK_MAP.get(ip);
if (lock != null) {
return lock;
}
synchronized (CREATION_LOCK) {
// 双重检查
lock = LOCK_MAP.get(ip);
if (lock == null) {
lock = new ReentrantLock(true); // 公平锁
LOCK_MAP.put(ip, lock);
}
return lock;
}
}
/**
* 安全加锁（带超时机制）
* @param ip IP地址
* @param timeout 超时时间
* @param unit 时间单位
* @return 是否成功获取锁
* @throws InterruptedException
*/
public static boolean lockWithTimeout(String ip, long timeout, TimeUnit unit)
throws InterruptedException {
ReentrantLock lock = getLockForIp(ip);
return lock.tryLock(timeout, unit);
}
/**
* 解锁指定IP的锁
* @param ip IP地址
*/
public static void unlockForIp(String ip) {
ReentrantLock lock = LOCK_MAP.get(ip);
if (lock != null && lock.isHeldByCurrentThread()) {
lock.unlock();
}
}
/**
* 清理未使用的锁（可选）
*/
public static void cleanupUnusedLocks() {
LOCK_MAP.entrySet().removeIf(entry ->
!entry.getValue().isLocked() &&
entry.getValue().getQueueLength() == 0
);
}
/**
* 获取当前锁数量
*/
public static int getLockCount() {
return LOCK_MAP.size();
}
/**
* 获取IP锁状态
* @param ip IP地址
* @return 锁状态字符串
*/
public static String getLockStatus(String ip) {
ReentrantLock lock = LOCK_MAP.get(ip);
if (lock == null) {
return "No lock exists";
}
return String.format("Locked: %s, Queue: %d, HoldCount: %d",
lock.isLocked(), lock.getQueueLength(), lock.getHoldCount());
}
/**
* 重置所有锁
*/
public static void reset() {
synchronized (CREATION_LOCK) {
LOCK_MAP.clear();
}
}
}
src/main/java/racal/sunyard/main/SydApi4j.java
+145 -31
View File
@@ -12,6 +12,7 @@ import com.sunyard.constant.CertUsage;
import com.sunyard.entity.Struct;
import com.sunyard.log.ILogFactory;
import com.sunyard.log.ILogger;
import com.sunyard.pool.ConnectionManager;
import com.sunyard.proto.Packet;
import com.sunyard.proto.PacketSection;
import com.sunyard.proto.Util;
@@ -55,6 +56,7 @@ import java.security.cert.X509Certificate;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Supplier;
import java.util.stream.Collectors;
@@ -69,6 +71,8 @@ public class SydApi4j implements SydApi {
Security.addProvider(new org.bouncycastle.jce.provider.BouncyCastleProvider());
}
private volatile boolean active;
private static boolean debug = false;
private static long sTime = System.currentTimeMillis();
@@ -78,6 +82,14 @@ public class SydApi4j implements SydApi {
private static int ha_port;
private static int ha_timeout;
private static String primaryIp;
private static String backupIp;
// 当前连接状态
private String currentActiveIp;
private static boolean usingBackup = false;
// 网络失败时重试次数
private String ip;
private int port;
@@ -117,6 +129,7 @@ public class SydApi4j implements SydApi {
* @param ip2
* @param port
* @param timeout
* todo 不应该这样使用
* @return
*/
public static int SYD_Set_Parameter_HA(
@@ -134,6 +147,34 @@ public class SydApi4j implements SydApi {
return 0;
}
// 心跳检测（可选实现）[1,6](@ref)
public void startHeartbeat() {
new Thread(() -> {
while (true) {
try {
Thread.sleep(5000); // 5秒检测一次
if (usingBackup) {
// 尝试恢复主节点
testPrimaryRecovery();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}).start();
}
// 主节点恢复检测
private void testPrimaryRecovery() {
try (Socket testSocket = new Socket()) {
testSocket.connect(new InetSocketAddress(ha_ip1, ha_port), 1000);
System.out.println("主节点恢复，准备切换...");
haConnect(); // 切换回主节点
} catch (IOException ignored) {
// 主节点尚未恢复
}
}
// 测试通过
@Override
public RetWrap SYD_SM2GenKeyPair() {
@@ -1353,25 +1394,15 @@ public class SydApi4j implements SydApi {
String result = "[" + failures.stream()
.map(Object::toString)
.collect(Collectors.joining(",")) + "]";
String message = "[SJJ1511] Certificate import failed - Servers:" + result + "| bkCode:[" + bkCode + "]";
String message = "支付签名证书导入失败！签名服务器地址：" + result + "| 证书对应行号：[" + bkCode + "]";
Gson gson = new Gson();
Map<String, String> map = gson.fromJson(snmpConfig, new TypeToken<Map<String, String>>() {}.getType());
try (SnmpTrapSender sender = new SnmpTrapSender(map)) {
ResponseEvent event = sender.sendTrap(message, 2);
sender.sendTrap(message, 2);
log.info("snmp发送告警消息:{}", message);
// if (event != null && event.getResponse() != null) {
// PDU response = event.getResponse();
// if (response.getErrorStatus() == PDU.noError) {
// log.info("snmp告警消息发送成功:{}/响应数据：{}", message, response.getVariableBindings());
// } else {
// log.info("snmp告警消息发送失败:{}", SnmpConstants.mpErrorMessage(response.getErrorStatus()));
// }
// }
} catch (Exception e) {
log.warn("snmp告警消息发送失败:{}", e.getMessage());
}
// throw new SydApiException("证书未全部导入成功！", 56300, true);
}
return dn;
}
@@ -1379,27 +1410,47 @@ public class SydApi4j implements SydApi {
private TaskContext<String> createTaskContext(String address, int timeout, String cert) {
Supplier<String> task = () -> importCertToServer(address, timeout, cert);
//重试2次，间隔1s
return new TaskContext<>(task, 2, Duration.ofSeconds(1), address);
return new TaskContext<>(task, 2, Duration.ofMillis(1000), address);
}
private String importCertToServer(String address, int timeout, String cert){
private String importCertToServer(String address, int timeout, String cert) {
// String dn = null;
// SydApi4j sydApi4j = null;
// try {
// sydApi4j = ConnectionManager.getInstance().getConnection(address, timeout);
// //导入证书
// dn = sydApi4j.importCertAndGetDN(cert);
// } finally {
// //保持长连接，暂时不清理
// if (sydApi4j != null) {
//// sydApi4j.disconnect();
// }
// }
// return dn;
//address 格式为 ip:port,拆分成ip和port，需要进行格式判断
String[] addressParts = address.split(":");
String ip = addressParts[0];
int port = Integer.parseInt(addressParts[1]);
String dn = null;
SydApi4j sydApi4j = null;
ReentrantLock ipLock = IpLockUtils.getLockForIp(address);
try {
//创建sydapi4j连接对象
sydApi4j = (SydApi4j) new SydApi4j().connect(ip, port, null, timeout);
//导入证书
dn = sydApi4j.importCertAndGetDN(cert);
ipLock.lock();
SydApi4j sydApi4j = null;
try {
//创建sydapi4j连接对象
sydApi4j = (SydApi4j) new SydApi4j().connect(ip, port, null, timeout);
//导入证书
dn = sydApi4j.importCertAndGetDN(cert);
} catch (Exception e) {
throw e;
} finally {
if (sydApi4j != null) {
sydApi4j.disconnect();
}
ipLock.unlock();
}
} catch (Exception e) {
throw e;
} finally {
if (sydApi4j != null) {
sydApi4j.disconnect();
}
}
return dn;
}
@@ -2216,6 +2267,9 @@ public class SydApi4j implements SydApi {
// }
// }
// }
if (ha_ip2 != null) {
startHeartbeat();
}
}
private Socket socket;
@@ -2280,8 +2334,11 @@ public class SydApi4j implements SydApi {
try {
s1.connect(address, ha_timeout);
socket = s1;
socket.setSoTimeout(ha_timeout);
currentActiveIp = ha_ip1;
usingBackup = false;
} catch (Exception e) {
Socket s2 = new Socket();
SocketAddress add2 = new InetSocketAddress(ha_ip2, ha_port);
@@ -2289,6 +2346,8 @@ public class SydApi4j implements SydApi {
s2.connect(add2, ha_timeout);
socket = s2;
socket.setSoTimeout(ha_timeout);
currentActiveIp = ha_ip2;
usingBackup = true;
} catch (Exception e2) {
throw new SydApiException(-2);
}
@@ -2337,9 +2396,17 @@ public class SydApi4j implements SydApi {
// 发送失败
if (null != ha_ip2) {
if (!usingBackup) {
// 主节点故障切换备机[3](@ref)
System.err.println("发送失败，切换到备机: " + ha_ip2);
haConnect(); // 重新连接备机
return syncSend(bb); // 重发数据
} else {
throw new SydApiException("主备节点均不可用", -10);
}
// HA 模式直接失败
e.printStackTrace();
throw new SydApiException("Socket write error", -2);
// e.printStackTrace();
// throw new SydApiException("Socket write error", -2);
} else {
// 非 HA 模式检查重试
socket = reConnect(new SydApiException("Socket write error", -2));
@@ -2358,6 +2425,22 @@ public class SydApi4j implements SydApi {
return buff;
}
private byte[] handleHaFailover(byte[] buff) {
System.err.println("主节点故障，切换到备用节点: " + ha_ip2);
try {
// 尝试连接备用节点
connect(ha_ip2, ha_port, null, ha_timeout);
// 重新发送数据
OutputStream os = socket.getOutputStream();
os.write(buff);
os.flush();
return buff;
} catch (Exception e) {
throw new SydApiException("HA切换失败: " + e.getMessage(), -6, e);
}
}
/**
* 开放此接口
*
@@ -2491,9 +2574,9 @@ public class SydApi4j implements SydApi {
@Override
synchronized public SydApi connect(String ip, int port, String pwd, int timeout) {
if (null != ha_ip2) {
throw new SydApiException("双机热备模式不用调用此方法", -1);
}
// if (null != ha_ip2) {
// throw new SydApiException("双机热备模式不用调用此方法", -1);
// }
Socket socket = new Socket();
SocketAddress address = new InetSocketAddress(ip, port);
@@ -2510,9 +2593,28 @@ public class SydApi4j implements SydApi {
this.socket = socket;
this.active = true;
return this;
}
// 智能连接主备节点
public synchronized void haConnect() {
// 优先连接主节点
try {
connect(ha_ip1, ha_port, null, ha_timeout);
currentActiveIp = ha_ip1;
usingBackup = false;
System.out.println("主节点连接成功: " + ha_ip1);
} catch (SydApiException e) {
// 主节点失败切换备机[2,5](@ref)
System.err.println("主节点故障，切换备机: " + ha_ip2);
connect(ha_ip2, ha_port, null, ha_timeout);
currentActiveIp = ha_ip2;
usingBackup = true;
}
}
@Override
public void init(SydApiBuilder builder) {
this.builder = builder;
@@ -2521,9 +2623,9 @@ public class SydApi4j implements SydApi {
@Override
synchronized public SydApi connect(String ip, int port, String pwd, int connectTimeout, int readTimeout) {
if (null != ha_ip2) {
throw new SydApiException("双机热备模式不用调用此方法", -1);
}
// if (null != ha_ip2) {
// throw new SydApiException("双机热备模式不用调用此方法", -1);
// }
Socket socket = new Socket();
SocketAddress address = new InetSocketAddress(ip, port);
@@ -2600,6 +2702,12 @@ public class SydApi4j implements SydApi {
throw e;
}
public boolean isActive() {
return active && socket != null &&
!socket.isClosed() &&
socket.isConnected();
}
@Override
synchronized public int disconnect() {
@@ -2607,18 +2715,24 @@ public class SydApi4j implements SydApi {
try {
socket.getInputStream().close();
} catch (Exception e1) {
} finally {
active = false;
}
try {
socket.getOutputStream().close();
} catch (Exception e2) {
} finally {
active = false;
}
try {
socket.close();
} catch (Exception e) {
} finally {
active = false;
}
}