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-06-04 10:13:37 +08:00
parent fbcbe79388
commit e4020b1bbc
4 changed files with 276 additions and 303 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
src/main/java/com/jmeter/test/Test4Jemter.java
View File

@ -49,6 +49,7 @@ public class Test4Jemter extends AbstractJavaSamplerClient {
arguments.addArgument("RSAdn", "C=cn,O=CFCA TEST CA,OU=NCS2,OU=Enterprises,CN=041@Z402451000010@ShanDong@00000001");
arguments.addArgument("orgData", "12345678");
arguments.addArgument("ip", "172.1.41.139");
arguments.addArgument("ip2", "172.1.41.96");
// arguments.addArgument("port", "8891");
// arguments.addArgument("ip2", "172.16.17.161");
arguments.addArgument("dataLength", "1024");
@ -69,6 +70,7 @@ public class Test4Jemter extends AbstractJavaSamplerClient {
RSAdn = context.getParameter("RSAdn");
oData = context.getParameter("orgData");
ip = context.getParameter("ip");
ip2 = context.getParameter("ip2");
// port = Integer.parseInt(context.getParameter("8891"));
// ip2 = context.getParameter("ip2");
dataLength = context.getIntParameter("dataLength");
@ -79,10 +81,12 @@ public class Test4Jemter extends AbstractJavaSamplerClient {
api = new ThreadLocal<SydApi4j>() {
@Override
public SydApi4j initialValue() {
SydApiBuilder builder = new SydApiBuilder().setIp(ip);
final SydApi4j api = (SydApi4j) builder.build();
return api;
//
// SydApiBuilder builder = new SydApiBuilder().setIp(ip);
// final SydApi4j api = (SydApi4j) builder.build();
SydApi4j.SYD_Set_Parameter_HA(ip, ip2, 8889, 1000);
final SydApi4j api1 = (SydApi4j) new SydApi4j();
return api1;
}
};

135
src/main/java/com/sunyard/pool/ConnectionManager.java
View File

@ -1,135 +0,0 @@
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();
}
}

367
src/main/java/racal/sunyard/main/SydApi4j.java
View File

@ -12,7 +12,6 @@ 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;
@ -33,7 +32,6 @@ import org.bouncycastle.jce.provider.BouncyCastleProvider;
import org.bouncycastle.pqc.math.linearalgebra.ByteUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.snmp4j.event.ResponseEvent;
import racal.sunyard.main.dto.DTO77;
import racal.sunyard.main.dto.DTO78;
import racal.sunyard.main.dto.DTO79;
@ -56,6 +54,8 @@ import java.security.cert.X509Certificate;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Supplier;
import java.util.stream.Collectors;
@ -71,24 +71,23 @@ 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();
/** --------------------------主备切换字段--------------------------------------------*/
private static String ha_ip1;
private static String ha_ip2;
private static int ha_port;
private static int ha_timeout;
private final AtomicBoolean usingBackup = new AtomicBoolean(false);
private static String primaryIp;
private static String backupIp;
private final Object reqLock = new Object();
// 当前连接状态
private String currentActiveIp;
private static boolean usingBackup = false;
private final Lock connectLock = new ReentrantLock();
/** --------------------------主备切换字段--------------------------------------------*/
// 网络失败时重试次数
private String ip;
@ -147,34 +146,6 @@ 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() {
@ -713,10 +684,12 @@ public class SydApi4j implements SydApi {
data.put("hash", hash);
ByteBuffer bb = proto.rend(packet);
synchronized(this) {
bb = this.syncRead(this.syncSend(bb));
bb = sendAndGetResponse(bb);
}
// synchronized(this) {
// bb = this.syncRead(this.syncSend(bb));
//
// }
Packet p = proto.parse(bb);
Map<String, Object> ret = p.toMap();
@ -736,6 +709,25 @@ public class SydApi4j implements SydApi {
}
}
private ByteBuffer sendAndGetResponse(ByteBuffer data) {
ByteBuffer readData = null;
int maxRetries = 2;
int retryCount = 0;
while (retryCount <= maxRetries) {
try {
synchronized(reqLock) {
byte[] sentData = this.syncSend(data);
readData = this.syncRead(sentData);
break;
}
} catch (SydApiException e) {
handleNetworkFailure(e, retryCount, maxRetries); // 统一处理网络故障
retryCount ++;
}
}
return readData;
}
public byte[] SYD_NakedSign_OuterHash(byte[] orgData, String sCertDN) {
// sCertDN = DnUtil.verifyDn(sCertDN);
String pk = null;
@ -1160,11 +1152,13 @@ public class SydApi4j implements SydApi {
// 渲染发送
ByteBuffer bb = proto.rend(packet);
synchronized (this) {
// 响应解析
bb = syncRead(syncSend(bb));
bb = sendAndGetResponse(bb);
}
// synchronized (this) {
// // 响应解析
// bb = syncRead(syncSend(bb));
//
// }
Packet p = proto.parse(bb);
Map<String, Object> ret = p.toMap();
if (
@ -1212,11 +1206,11 @@ public class SydApi4j implements SydApi {
// 渲染发送
ByteBuffer bb = proto.rend(packet);
synchronized (this) {
// 响应解析
bb = syncRead(syncSend(bb));
}
// synchronized (this) {
// // 响应解析
// bb = syncRead(syncSend(bb));
// }
bb = sendAndGetResponse(bb);
Packet p = proto.parse(bb, packet);
@ -1414,19 +1408,6 @@ public class SydApi4j implements SydApi {
}
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];
@ -1439,6 +1420,8 @@ public class SydApi4j implements SydApi {
try {
//创建sydapi4j连接对象
sydApi4j = (SydApi4j) new SydApi4j().connect(ip, port, null, timeout);
//先查询
//导入证书
dn = sydApi4j.importCertAndGetDN(cert);
} catch (Exception e) {
@ -2156,6 +2139,17 @@ public class SydApi4j implements SydApi {
public boolean deleteCert(String dn) {
Proto8011 proto = new Proto8011();
byte[] dnData = new byte[256];
byte[] s = null;
try {
// sCertDN = verifyDn(sCertDN);
s = dn.getBytes("GBK");
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
System.arraycopy(s, 0, dn, 0, s.length);
// 填充数据
HashMap<String, Object> packet = new HashMap<String, Object>();
@ -2164,7 +2158,7 @@ public class SydApi4j implements SydApi {
PacketSN sn = PacketSN.gen();
packet.put("header", sn.getSn().array());
data.put("certDn", dn);
data.put("certDn", dnData);
@ -2267,9 +2261,6 @@ public class SydApi4j implements SydApi {
// }
// }
// }
if (ha_ip2 != null) {
startHeartbeat();
}
}
private Socket socket;
@ -2328,30 +2319,29 @@ public class SydApi4j implements SydApi {
// 全局热备模式（old shool）
if (null != ha_ip2) {
Socket s1 = new Socket();
SocketAddress address = new InetSocketAddress(ha_ip1, ha_port);
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);
try {
s2.connect(add2, ha_timeout);
socket = s2;
socket.setSoTimeout(ha_timeout);
currentActiveIp = ha_ip2;
usingBackup = true;
} catch (Exception e2) {
throw new SydApiException(-2);
}
}
connectHa();
//
// Socket s1 = new Socket();
// SocketAddress address = new InetSocketAddress(ha_ip1, ha_port);
//
// try {
// s1.connect(address, ha_timeout);
//
// socket = s1;
// socket.setSoTimeout(ha_timeout);
// usingBackup.set(false);
// } catch (Exception e) {
// Socket s2 = new Socket();
// SocketAddress add2 = new InetSocketAddress(ha_ip2, ha_port);
// try {
// s2.connect(add2, ha_timeout);
// socket = s2;
// socket.setSoTimeout(ha_timeout);
// usingBackup.set(true);
// } catch (Exception e2) {
// throw new SydApiException(-2);
// }
// }
}
}
@ -2360,6 +2350,96 @@ public class SydApi4j implements SydApi {
}
/**
* 连接或重连到有效节点
*/
public synchronized void connectHa() {
// 优先尝试主节点
if (!usingBackup.get()) {
if (tryConnect(ha_ip1)) {
System.out.println("连接到主节点: " + ha_ip1);
return;
}
}
// 连接到备节点
if (tryConnect(ha_ip2)) {
System.out.println("连接到备节点: " + ha_ip2);
usingBackup.set(true);
return;
}
throw new SydApiException("所有节点均不可用", -100);
}
/**
* 尝试连接到指定节点
*/
private boolean tryConnect(String ip) {
try {
Socket newSocket = new Socket();
InetSocketAddress address = new InetSocketAddress(ip, ha_port);
if (ha_timeout > 0) {
newSocket.connect(address, ha_timeout);
} else {
newSocket.connect(address);
}
newSocket.setSoTimeout(ha_timeout);
// 关闭旧socket（如果存在）
safeClose();
this.socket = newSocket;
return true;
} catch (IOException e) {
System.out.println("节点连接失败: " + ip + " - " + e.getMessage());
return false;
}
}
private synchronized void handleNetworkFailure(Exception e, int retryCount, int maxRetries) {
if (retryCount > maxRetries) {
System.out.println(Thread.currentThread().getName() + "重试次数耗尽");
return;
}
System.out.printf("线程[%s]网络故障，尝试切换节点 (重试: %d/%d)%n",
Thread.currentThread().getName(), retryCount + 1, maxRetries);
boolean currentState = usingBackup.get();
boolean newState = !currentState;
if (usingBackup.compareAndSet(currentState, newState)) {
String targetIp = newState ? ha_ip2 : ha_ip1;
System.out.println("全局节点切换: " + (currentState ? "备机" : "主机") + "→" +
(newState ? "备机" : "主机"));
try {
safeClose();
connect(targetIp, ha_port, ha_timeout);
System.out.println("线程[" + Thread.currentThread().getName() + "]连接重建成功");
} catch (Exception ex) {
System.out.println("线程[" + Thread.currentThread().getName() + "]节点切换失败: " + ex.getMessage());
usingBackup.compareAndSet(newState, currentState);
}
} else {
System.out.println("线程[" + Thread.currentThread().getName() + "]节点已被其他线程切换，使用新节点");
}
}
private void safeClose() {
if (socket != null) {
try {
socket.close();
} catch (IOException e) {
System.out.println("关闭连接异常: " + e.getMessage());
}
socket = null;
}
}
protected byte[] syncSend(ByteBuffer bb) {
@ -2396,17 +2476,9 @@ 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));
@ -2425,21 +2497,6 @@ 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);
}
}
/**
* 开放此接口
@ -2465,13 +2522,15 @@ public class SydApi4j implements SydApi {
private void clear() {
try {
InputStream is = socket.getInputStream();
while (is.available() > 0) {
is.read();
if (null != socket) {
try {
InputStream is = socket.getInputStream();
while (is.available() > 0) {
is.read();
}
} catch (Exception e) {
e.printStackTrace();
}
} catch (Exception e) {
e.printStackTrace();
}
}
@ -2571,6 +2630,24 @@ public class SydApi4j implements SydApi {
throw new SydApiException("通信数据错误", -3);
}
private void connect(String ip, int port, int timeout) {
synchronized (reqLock) {
safeClose();
try {
Socket newSocket = new Socket();
SocketAddress address = new InetSocketAddress(ip, port);
newSocket.connect(address, timeout);
newSocket.setSoTimeout(timeout);
this.socket = newSocket;
System.out.printf("线程[%s]连接到%s:%d%n",
Thread.currentThread().getName(), ip, port);
} catch (IOException e) {
throw new SydApiException("连接失败: " + ip + ":" + port, -2, e);
}
}
}
@Override
synchronized public SydApi connect(String ip, int port, String pwd, int timeout) {
@ -2593,28 +2670,10 @@ 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;
@ -2702,12 +2761,6 @@ public class SydApi4j implements SydApi {
throw e;
}
public boolean isActive() {
return active && socket != null &&
!socket.isClosed() &&
socket.isConnected();
}
@Override
synchronized public int disconnect() {
@ -2716,7 +2769,6 @@ public class SydApi4j implements SydApi {
socket.getInputStream().close();
} catch (Exception e1) {
} finally {
active = false;
}
@ -2724,7 +2776,6 @@ public class SydApi4j implements SydApi {
socket.getOutputStream().close();
} catch (Exception e2) {
} finally {
active = false;
}
@ -2732,7 +2783,6 @@ public class SydApi4j implements SydApi {
socket.close();
} catch (Exception e) {
} finally {
active = false;
}
}
@ -7255,10 +7305,12 @@ public class SydApi4j implements SydApi {
bb.put(1, (byte) (0xFF & (len % 256)));
synchronized (this) {
// 响应解析
bb = syncRead(syncSend(bb));
}
// synchronized (this) {
// // 响应解析
// bb = syncRead(syncSend(bb));
// }
bb = sendAndGetResponse(bb);
bb.flip();
@ -8094,11 +8146,12 @@ public class SydApi4j implements SydApi {
// 渲染发送
ByteBuffer bb = proto.rend(packet);
//System.out.println(Util.bytes2HexString(Util.toArray(bb)));
synchronized (this) {
// 响应解析
bb = syncRead(syncSend(bb));
}
// synchronized (this) {
//
// // 响应解析
// bb = syncRead(syncSend(bb));
// }
bb = sendAndGetResponse(bb);
//ProtocolParser parser = pset.getProtocolResParse();
Packet p = proto.parse(bb);
Map<String, Object> ret = p.toMap();

65
src/test/java/sample/Test4NongxinCW.java
View File

@ -15,6 +15,7 @@ import racal.sunyard.main.SydApi4j;
import java.security.cert.X509Certificate;
import java.util.*;
import java.util.concurrent.*;
import static com.sunyard.util.CertUtil.convertToX509Cert;
@ -22,19 +23,19 @@ public class Test4NongxinCW {
private SydApi4j api;
final String SM2dn = "C=CN,ST=zz,L=zz,O=zzz,CN=zzzzz";
final String SM2dn = "1C=CN,ST=zz,L=zz,O=zzz,CN=zzzzz";
byte[] orgData = new byte[100];
@Before
public void before() {
//主备调用接口
SydApi4j.SYD_Set_Parameter_HA("172.1.41.139", "172.1.41.96", 8889, 1000);
SydApi4j.SYD_Set_Parameter_HA("172.1.41.96", "172.1.41.139", 8889, 1000);
api = (SydApi4j) new SydApi4j();
api.startHeartbeat();
// api.startHeartbeat();
// 通过设置debug的属性开启debug，如果不设置则默认不打开
api = (SydApi4j) new SydApi4j().connect("172.1.41.139", 8891, null, 1000);
// api = (SydApi4j) new SydApi4j().connect("172.1.41.139", 8891, null, 1000);
}
@ -52,6 +53,56 @@ public class Test4NongxinCW {
}
}
@Test
public void testHA() {
for (int i = 0; i < 10000; i++) {
byte[] nakedSign = api.SYD_NakedSign(orgData,SM2dn);
}
}
@Test
public void tes() throws InterruptedException, ExecutionException {
// 配置HA参数
// SydApi4j.SYD_Set_Parameter_HA("172.1.41.139", "172.1.41.96", 8889, 1000);
// 创建20个线程并发操作API
ExecutorService executor = Executors.newFixedThreadPool(20);
List<Future<?>> futures = new ArrayList<>();
// 模拟主节点故障
CountDownLatch startLatch = new CountDownLatch(1);
for (int i = 0; i < 20; i++) {
futures.add(executor.submit(() -> {
try {
startLatch.await(); // 等待统一开始
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
SydApi4j api2 = new SydApi4j();
for (int j = 0; j < 10000; j++) {
byte[] nakedSign = api2.SYD_NakedSign(orgData,SM2dn);
}
}));
}
// 触发所有线程开始
startLatch.countDown();
// 随机触发主节点故障（模拟网络中断）
Thread.sleep(500);
// 等待所有任务完成
for (Future<?> future : futures) {
future.get();
}
executor.shutdown();
}
@Test
public void nakeSign(){
@ -293,9 +344,9 @@ public class Test4NongxinCW {
@Test
public void selectCert() throws Exception {
// String SM2dn1 = "C=CN,ST=1,L=1,O=1,OU=1,OU=1,CN=1";
// String s = api.getCertByDn(SM2dn);
// System.out.println(s);
String SM2dn1 = "C=CN,ST=1,L=1,O=1,OU=1,OU=1,CN=1";
String s = api.getCertByDn(SM2dn);
System.out.println(s);
String message = "sdas发顺丰";
byte[] gbkBytes = message.getBytes("GBK");
System.out.println("GBK Encoded Bytes: " + Arrays.toString(gbkBytes));