github/chinaport-data-signature
4
0
Fork 0
mirror of https://github.com/Weasley-J/chinaport-data-signature.git synced 2026-10-19 16:15:47 +00:00
Code Issues Projects Releases Wiki Activity

Merge branch 'feature_x' into feature_download_x509_certificate

Browse Source
This commit is contained in:
julian
2026-09-07 22:30:08 +08:00
parent 54c34fa9a0 2b4d18eb3a
commit ca999da5a9
16 changed files with 804 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+1
View File
@@ -241,6 +241,7 @@ _参见[start.sh](https://github.com/julxxy/chinaport-data-signature/blob/main/s
- 小红书
- 广东铭鸿数据有限公司
- 海口中伟机免投资有限公司
- 远方好物跨境供应链（深圳）有限公司 [官网](https://www.yuanfangkuajing.com/home)
- ...
---
chinaport-data-signature-webapp/docker_build_push_multiarch.sh
+10 -6
View File
@@ -2,21 +2,25 @@
# 检查并安装 sptn 命令
source ~/.zshrc
if command -v sptn >/dev/null 2>&1; then
echo "检测到 sptn 命令，正在执行..."
sptn
if command -v setproxy >/dev/null 2>&1; then
echo "检测到 setproxy 命令，正在执行..."
setproxy
else
echo "未检测到 sptn 命令，跳过执行"
echo "未检测到 setproxy 命令，跳过执行"
fi
export JAVA_HOME=$(/usr/libexec/java_home -v 17)
export PATH="$JAVA_HOME/bin:$PATH"
echo "当前 JDK 版本: $(java -version 2>&1 | head -1)"
# --- 进入项目目录 ---
cd ..
SETTINGS="/Users/weasley/Development/program/apache-maven/conf/settings-aliyun.xml"
SETTINGS="/Users/${USER}/Development/program/apache-maven/conf/settings-aliyun.xml"
mvn clean package --settings $SETTINGS
cd chinaport-data-signature-webapp
# --- 配置信息 ---
TAG="1.2.0.1"
TAG="1.2.1"
IMAGE="weasleyj/chinaport-data-signature"
CONTAINER_NAME="chinaport-data-signature"
BUILDER_NAME="multiarch-builder"
chinaport-data-signature-webapp/pom.xml
+1 -1
View File
@@ -11,7 +11,7 @@
<groupId>cn.alphahub.eport.signature</groupId>
<artifactId>chinaport-data-signature-webapp</artifactId>
<version>1.2.0</version>
<version>1.2.1</version>
<name>chinaport-data-signature-webapp</name>
<url>https://github.com/julxxy/chinaport-data-signature.git</url>
chinaport-data-signature-webapp/src/main/java/cn/alphahub/eport/signature/controller/rpc/EportReportResultController.java
+16
View File
@@ -87,4 +87,20 @@ public class EportReportResultController {
return Result.ok(msgResult.block());
}
/**
* 查询报错信息
*
* @param qryId 查询时间，日期时间格式：yyyyMMddHHmmss
* @return 报错信息
*/
@GetMapping("/ceb900msg")
public Result<String> getCeb900msgResult(@RequestParam(required = false) @DateTimeFormat(pattern = "yyyyMMddHHmmss") LocalDateTime qryId) {
qryId = ObjectUtils.getIfNull(qryId, LocalDateTime.now().minusSeconds(15));
log.info("开始查询报错信息, qryId {}", FORMATTER.format(qryId));
String traceId = MDC.get(TRACE_ID);
WebClientConfiguration.HTTP_HEADERS.set(h -> h.add(TRACE_ID, traceId));
Mono<String> msgResult = eportReportResultHttpClient.getCeb900msgResult(chinaEportProperties.getDxpId(), FORMATTER.format(qryId))
.contextWrite(Context.of(TRACE_ID, traceId));
return Result.ok(msgResult.block());
}
}
chinaport-data-signature-webapp/src/main/java/cn/alphahub/eport/signature/core/SignHandler.java
+2
View File
@@ -179,6 +179,8 @@ public class SignHandler {
}
} catch (Exception e) {
log.error("唤醒线程异常 {}", e.getLocalizedMessage(), e);
} finally {
LockSupport.unpark(ref.get().getThread());
}
}
chinaport-data-signature-webapp/src/main/java/cn/alphahub/eport/signature/core/web/EportReportResultHttpClient.java
+11
View File
@@ -55,4 +55,15 @@ public interface EportReportResultHttpClient {
*/
@GetExchange(value = "/ceb622msg", accept = {MediaType.TEXT_PLAIN_VALUE})
Mono<String> getCe622msgResult(@RequestParam String dxpid, @RequestParam String qryid);
/**
* 报错信息
*
* @param dxpid 向中国电子口岸数据中心申请数据交换平台的用户编号
* @param qryid 查询时间，日期时间格式：yyyyMMddHHmmss
* @return 报错信息
* @apiNote 查询下 311，621 等申报回执时，如果出现报错信息，可以通过此接口查询报错信息
*/
@GetExchange(value = "/ceb900msg", accept = {MediaType.TEXT_PLAIN_VALUE})
Mono<String> getCeb900msgResult(@RequestParam String dxpid, @RequestParam String qryid);
}
chinaport-data-signature-webapp/src/main/java/cn/alphahub/eport/signature/support/CertificateParser.java
+1 -1
View File
@@ -34,7 +34,7 @@ public final class CertificateParser {
/**
* 根据 'xxx.cer' 证书获取 X509 证书
*
* @param filePath 文件名， 如：/Users/weasley/Downloads/01691fe9.cer
* @param filePath 文件名， 如：/Users/julian/Downloads/01691fe9.cer
* @return X.509 certificate
*/
@SuppressWarnings("all")
chinaport-data-signature-webapp/src/test/java/cn/alphahub/eport/signature/controller/rpc/EportSignControllerTest.java
+22 -6
View File
@@ -4,6 +4,7 @@ import cn.alphahub.eport.signature.config.ChinaEportProperties;
import cn.alphahub.eport.signature.core.ChinaEportReportClient;
import cn.alphahub.eport.signature.core.SignHandler;
import cn.alphahub.eport.signature.core.web.EportCebMessageHttpClient;
import cn.alphahub.eport.signature.core.web.EportReportResultHttpClient;
import cn.alphahub.eport.signature.entity.SignRequest;
import cn.alphahub.eport.signature.entity.SignResult;
import cn.alphahub.eport.signature.entity.UkeyRequest;
@@ -34,6 +35,7 @@ import org.springframework.boot.test.context.SpringBootTest;
@Slf4j
@SpringBootTest
class EportSignControllerTest {
@Autowired
SignHandler signHandler;
@Autowired
@@ -44,6 +46,8 @@ class EportSignControllerTest {
ChinaEportReportClient chinaEportReportClient;
@Autowired
EportCebMessageHttpClient cebMessageHttpClient;
@Autowired
EportReportResultHttpClient eportReportResultHttpClient;
@Test
@DisplayName("海关XML数据加签+验正签名结果")
@@ -151,6 +155,7 @@ class EportSignControllerTest {
ceb311Message.getOrder().getOrderHead().setGuid(guid);
chinaEportReportClient.buildBaseTransfer(ceb311Message.getBaseTransfer());
MessageRequest messageRequest = chinaEportReportClient.buildMessageRequest(ceb311Message, MessageType.CEB311Message);
/*
String requestServer = StringUtils.defaultIfBlank(chinaEportProperties.getServer(), Base64.decodeStr(EPORT_CEBMESSAGE_SERVER_ENCODE));
String requestBody = JSONUtil.toJsonStr(messageRequest);
@@ -161,22 +166,33 @@ class EportSignControllerTest {
.execute();
log.info("开始上报，Http响应结果: {}", httpResponse.body());
*/
cebMessageHttpClient.reportCebMessage(messageRequest).subscribe(data -> {
log.info("开始申报数据，Http响应结果: {}", data);
});
// 睡8秒，开始查询回执
// 睡8秒，开始查询回执、报错信息
LocalDateTime uploadTime = LocalDateTime.now();
int sleepTime = 15;
for (int i = sleepTime; i > 0; i--) {
String currDate = DateTimeFormatter.ofPattern("yyyyMMddHHmmss").format(uploadTime);
for (int i = 15; i > 0; i--) {
log.warn("开始查询推送回执，海关后台正在处理中，线程挂起剩余 {} 秒。", i);
TimeUnit.SECONDS.sleep(1);
}
String url = "http://" + Base64.decodeStr("MzYuMTAxLjIwOC4yMzA=") + ":8090/ceb312msg?" + HttpUtil.toParams(new LinkedHashMap<>() {{
String url = "http://" + Base64.decodeStr("MzYuMTAxLjIwOC4yMzA=") + ":8090/ceb900msg?" + HttpUtil.toParams(new LinkedHashMap<>() {{
put("dxpid", chinaEportProperties.getDxpId());
put("qryid", DateTimeFormatter.ofPattern("yyyyMMddHHmmss").format(uploadTime));
put("qryid", currDate);
}});
HttpResponse receiptResponse = HttpUtil.createGet(url).execute();
log.info("\n{}\n回执结果: {}", url, receiptResponse.body()); //回执结果: [{org_id=1, instm=1690556389000, re_order_guid=AOF0YM-WEASLEY-20230728225932-6QVICD, returnstatus=-301014, ebpcode=4601630004, returntime2=20230728225949, ebccode=4601630004, trans_dxpid=DXPENT0000530815, returninfo=订单报文处理失败，接收报文为旧报文数据，不做处理;, returntime=1690556377000, orderno=T_C5051511332138160010}]
log.info("\n{}\n900回执结果: {}", url, receiptResponse.body()); //
url = "http://" + Base64.decodeStr("MzYuMTAxLjIwOC4yMzA=") + ":8090/ceb312msg?" + HttpUtil.toParams(new LinkedHashMap<>() {{
put("dxpid", chinaEportProperties.getDxpId());
put("qryid", currDate);
}});
receiptResponse = HttpUtil.createGet(url).execute();
log.info("\n{}\n311回执结果: {}", url, receiptResponse.body()); //回执结果: [{org_id=1, instm=1690556389000, re_order_guid=AOF0YM-WEASLEY-20230728225932-6QVICD, returnstatus=-301014, ebpcode=4601630004, returntime2=20230728225949, ebccode=4601630004, trans_dxpid=DXPENT0000530815, returninfo=订单报文处理失败，接收报文为旧报文数据，不做处理;, returntime=1690556377000, orderno=T_C5051511332138160010}]
}
}
chinaport-data-signature-webapp/src/test/java/cn/alphahub/eport/signature/core/CertificateHandlerParserTest.java
+2 -2
View File
@@ -17,7 +17,7 @@ class CertificateHandlerParserTest {
@Test
void parseCertificateByCertPath() {
CertificateParser.parseCertificateByCertPath("/Users/weasley/Downloads/03000000000cde6f.cer");
CertificateParser.parseCertificateByCertPath("/Users/weasley/Downloads/01691fe9.cer");
CertificateParser.parseCertificateByCertPath("/Users/julian/Downloads/03000000000cde6f.cer");
CertificateParser.parseCertificateByCertPath("/Users/julian/Downloads/01691fe9.cer");
}
}
chinaport-data-signature-webapp/src/test/java/cn/alphahub/eport/signature/core/PdfFileSignatureTests.java
+116
View File
@@ -0,0 +1,116 @@
package cn.alphahub.eport.signature.core;
import cn.alphahub.eport.signature.entity.UkeyRequest;
import cn.alphahub.eport.signature.entity.UkeyResponse.Args;
import cn.hutool.core.codec.Base64;
import cn.hutool.json.JSONUtil;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.LinkedHashMap;
import java.util.Map;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
/**
* PDF 签名.
*
* @version 1.0.0
* @date 8/15/26 16:20
*/
@Slf4j
@SpringBootTest
class PdfFileSignatureTests {
/**
* 实际下发的 ukey 方法
*/
static final String PDF_ADD_SIGN_METHOD = "pdf-sign_PdfAddSign";
/**
* 实际下发的 ukey 方法
*/
static final String PDF_CHECK_SIGN_METHOD = "pdf-sign_PdfCheckSign";
/**
* 随附单据类型编码，按实际业务调整
*/
static final String EDOC_CODE = "00000001";
/**
* u-key默认密码8个8
*/
private static final String DEFAULT_PASSWORD = "88888888";
/**
* 根据实际情况替换
*/
final String PDF_PATH = "doc/sign-pdf-test.pdf";
@Autowired
CertificateHandler certHandler;
@Autowired
SignHandler signHandler;
@Test
@DisplayName("海关PDF文件加签+验证签名结果")
void signPdfFile() {
Path pdfFile = resolvePdfFile();
String destDir = pdfFile.getParent().toString();
String fileName = pdfFile.getFileName().toString();
String datetiem = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMddHHmmss"));
Map<String, Object> signArgsMap = new LinkedHashMap<>();
signArgsMap.put("passwd", DEFAULT_PASSWORD);
signArgsMap.put("filePath", destDir);
signArgsMap.put("fileName", fileName);
signArgsMap.put("datetiem", datetiem);
signArgsMap.put("Edoc_code", EDOC_CODE);
Args signArgs = signHandler.getUkeyResponseArgs(new UkeyRequest(PDF_ADD_SIGN_METHOD, signArgsMap));
System.err.println("PDF加签结果: " + JSONUtil.toJsonStr(signArgs));
System.err.println("X509Certificate: " + certHandler.getX509Certificate(CertificateHandler.SING_DATA_METHOD));
Map<String, Object> verifyArgsMap = new LinkedHashMap<>();
verifyArgsMap.put("filePath", pdfFile.toString());
Args verifyArgs = signHandler.getUkeyResponseArgs(new UkeyRequest(PDF_CHECK_SIGN_METHOD, verifyArgsMap));
System.err.println("PDF验签结果: " + JSONUtil.toJsonStr(verifyArgs));
}
@Test
@DisplayName("海关PDF Base64内容加签+验证签名结果")
void signPdfFileContent() throws Exception {
Path pdfFile = resolvePdfFile();
String fileContent = Base64.encode(Files.readAllBytes(pdfFile));
String datetiem = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMddHHmmss"));
Map<String, Object> signArgsMap = new LinkedHashMap<>();
signArgsMap.put("fileContent", fileContent);
signArgsMap.put("datetiem", datetiem);
signArgsMap.put("Edoc_code", EDOC_CODE);
signArgsMap.put("passwd", DEFAULT_PASSWORD);
Args signArgs = signHandler.getUkeyResponseArgs(new UkeyRequest(PDF_ADD_SIGN_METHOD, signArgsMap));
System.err.println("PDF加签结果: " + JSONUtil.toJsonStr(signArgs));
System.err.println("X509Certificate: " + certHandler.getX509Certificate(CertificateHandler.SING_DATA_METHOD));
String signedPdfBase64 = signArgs.getData().isEmpty() ? fileContent : signArgs.getData().get(0);
Map<String, Object> verifyArgsMap = new LinkedHashMap<>();
verifyArgsMap.put("fileContent", signedPdfBase64);
Args verifyArgs = signHandler.getUkeyResponseArgs(new UkeyRequest(PDF_CHECK_SIGN_METHOD, verifyArgsMap));
System.err.println("PDF验签结果: " + JSONUtil.toJsonStr(verifyArgs));
}
private Path resolvePdfFile() {
Path fromCwd = Path.of(PDF_PATH);
if (Files.isRegularFile(fromCwd)) {
return fromCwd.toAbsolutePath().normalize();
}
Path fromModuleParent = Path.of("..").resolve(PDF_PATH);
if (Files.isRegularFile(fromModuleParent)) {
return fromModuleParent.toAbsolutePath().normalize();
}
throw new IllegalStateException("未找到测试 PDF: " + PDF_PATH);
}
}
chinaport-data-signature-webapp/src/test/java/cn/alphahub/eport/signature/core/v121/MySignHandler.java
+136
View File
@@ -0,0 +1,136 @@
package cn.alphahub.eport.signature.core.v121;
import cn.alphahub.eport.signature.base.constant.FrameworkConstant;
import cn.alphahub.eport.signature.config.UkeyInitialConfig;
import cn.alphahub.eport.signature.config.UkeyProperties;
import cn.alphahub.eport.signature.core.SignatureHandler;
import cn.alphahub.eport.signature.core.WebSocketClientHandler;
import cn.alphahub.eport.signature.core.notify.EmailNotifyStrategy;
import cn.alphahub.eport.signature.entity.SignRequest;
import cn.alphahub.eport.signature.entity.SignResult;
import cn.alphahub.eport.signature.entity.WebSocketWrapper;
import cn.alphahub.eport.signature.support.XMLValidator;
import jakarta.annotation.PreDestroy;
import jakarta.validation.Valid;
import jakarta.validation.constraints.NotBlank;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.LockSupport;
import lombok.Data;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.Strings;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.socket.client.WebSocketConnectionManager;
import org.springframework.web.socket.client.standard.StandardWebSocketClient;
/**
* 电子口岸加签业务核心类（最终优化版）
* 核心：仅解决大消息1009错误 + 连接复用，业务逻辑与原SignHandler完全一致
*/
@Data
@Slf4j
//@Service("mySignHandler") // 唯一Bean名，避免与原SignHandler冲突
@Validated
@Accessors(chain = true)
public class MySignHandler {
// 原逻辑常量（完全复制）
private static final long LOCK_TIMEOUT_NANOS = 1000 * 1000 * 1000 * 5L;
// 大消息配置（解决1009错误）
private static final int MAX_TEXT_MESSAGE_SIZE = 1024 * 1024; // 1MB
@Autowired
private UkeyProperties ukeyProperties;
@Autowired
private WebSocketClientHandler webSocketClientHandler; // 复用原全局Handler
@Autowired
private StandardWebSocketClient standardWebSocketClient;
@Autowired
@Qualifier(EmailNotifyStrategy.NAME)
private EmailNotifyStrategy emailNotifyStrategy;
// 连接复用（仅这一个优化点）
private WebSocketConnectionManager reuseConnManager;
// -------------------- 业务方法完全复制原SignHandler --------------------
public static String getInitData(@Valid SignRequest request) {
if (isSignXml(request)) {
return SignatureHandler.getSignatureNodeBeforeSend(request);
}
return request.getData();
}
public static boolean isSignXml(@Valid SignRequest request) {
if (StringUtils.isBlank(request.getData())) {
throw new IllegalArgumentException("加签数据请求入参能为空!");
}
boolean isSignXmlString = XMLValidator.isValidXML(request.getData()) || Strings.CS.startsWith(request.getData(), "<ceb:CEB");
boolean isSign179String = Strings.CS.startsWith(request.getData(), "\"sessionID\"");
if (isSignXmlString && !isSign179String) {
return true;
}
if (isSign179String && !isSignXmlString) {
return false;
}
throw new IllegalArgumentException("加签数据请求入参不合法,请检查参数:" + request.getData());
}
public String getSignDataParameter(@Valid SignRequest request) {
return UkeyInitialConfig.getSignDataAsPEM(request);
}
/**
* 核心sign方法：完全复制原逻辑，仅优化连接复用
*/
public SignResult sign(@Valid SignRequest request, @NotBlank(message = "websocket发送的数据载荷不能为空") String payload) {
log.info("收到u-key加签数据: {}", payload);
// 完全复制原逻辑：创建WebSocketWrapper
WebSocketWrapper wsWrapper = webSocketClientHandler.getWebSocketWrapper();
wsWrapper.setPayload(payload);
wsWrapper.setRequest(request);
wsWrapper.setSignResult(new SignResult());
wsWrapper.setThreadReference(new AtomicReference<>(Thread.currentThread()));
wsWrapper.setSessionId(MDC.get(FrameworkConstant.TRACE_ID));
// 完全复制原逻辑：XML加签字段赋值
if (MySignHandler.isSignXml(request)) {
wsWrapper.getSignResult().setDigestValue(SignatureHandler.getDigestValueOfCEBXxxMessage(request.getData()));
wsWrapper.getSignResult().setSignatureNode(SignatureHandler.getSignatureNodeBeforeSend(request));
}
// 优化点：连接复用（替代原每次new）
if (reuseConnManager == null) {
reuseConnManager = new WebSocketConnectionManager(
standardWebSocketClient, webSocketClientHandler, ukeyProperties.getWsUrl()
);
}
WebSocketConnectionManager wsConnManager = reuseConnManager;
// 完全复制原逻辑：启动连接 + 阻塞等待
wsConnManager.start();
try {
LockSupport.parkNanos(wsWrapper.getThreadReference().get(), LOCK_TIMEOUT_NANOS);
} catch (Exception e) {
log.error("线程自动unpark异常 {}", e.getLocalizedMessage(), e);
wsWrapper.getSignResult().setSuccess(false);
}
// 完全复制原逻辑：返回结果（signatureValue由原WebSocketClientHandler赋值）
return wsWrapper.getSignResult();
}
// 销毁资源：关闭复用的连接
@PreDestroy
public void destroy() {
if (reuseConnManager != null && reuseConnManager.isRunning()) {
reuseConnManager.stop();
}
log.info("MySignHandler资源已释放");
}
}
chinaport-data-signature-webapp/src/test/java/cn/alphahub/eport/signature/core/v121/UkeySignService.java
+469
View File
@@ -0,0 +1,469 @@
package cn.alphahub.eport.signature.core.v121;
import cn.alphahub.eport.signature.config.UkeyProperties;
import cn.alphahub.eport.signature.core.notify.EmailNotifyStrategy;
import cn.alphahub.eport.signature.core.notify.model.EmailNotifyRecord;
import cn.alphahub.eport.signature.entity.SignRequest;
import cn.alphahub.eport.signature.entity.SignResult;
import cn.alphahub.eport.signature.entity.UkeyRequest;
import cn.alphahub.eport.signature.entity.UkeyResponse;
import cn.alphahub.eport.signature.entity.WebSocketWrapper;
import cn.hutool.json.JSONUtil;
import jakarta.annotation.PreDestroy;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import lombok.Data;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.lang.NonNull;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.client.WebSocketConnectionManager;
import org.springframework.web.socket.client.standard.StandardWebSocketClient;
import org.springframework.web.socket.handler.TextWebSocketHandler;
/**
* U-Key加签服务最终兼容版（适配所有Spring版本，移除getSession()依赖）
*/
@Data
@Slf4j
//@Service
@Validated
@Accessors(chain = true)
public class UkeySignService {
private static final String TRACE_ID = FrameworkConstant.TRACE_ID;
private static final long CONNECT_TIMEOUT_MS = 5000; // 连接超时
private static final long RESPONSE_TIMEOUT_MS = 3000; // 响应超时
private static final long HEARTBEAT_INTERVAL_SEC = 30; // 心跳间隔
private static final long RECONNECT_RETRY_DELAY_MS = 1000; // 重连重试间隔
private static final int RECONNECT_MAX_RETRY = 5; // 重连最大重试次数
// 核心组件
private final WebSocketConnectionManager wsConnManager;
private final AtomicReference<WebSocketSession> sessionRef = new AtomicReference<>(); // 缓存Session
private final Map<String, CompletableFuture<UkeyResponse.Args>> requestFutureMap = new ConcurrentHashMap<>();
private final ScheduledExecutorService heartbeatExecutor = Executors.newSingleThreadScheduledExecutor();
private final AtomicBoolean isConnected = new AtomicBoolean(false);
private final Object connectLock = new Object();
// 配置常量（建议通过配置文件注入）
@Autowired
private UkeyProperties ukeyProperties;
/**
* 电子口岸u-key加签失败通知策略
*/
@Autowired
@Qualifier(EmailNotifyStrategy.NAME)
private EmailNotifyStrategy emailNotifyStrategy;
// 初始化
public UkeySignService(UkeyProperties ukeyProperties, EmailNotifyStrategy emailNotifyStrategy) {
this.ukeyProperties = ukeyProperties;
this.emailNotifyStrategy = emailNotifyStrategy;
this.wsConnManager = createWebSocketManager();
startHeartbeat(); // 启动心跳
}
/**
* 创建WebSocket管理器（完全兼容低版本Spring）
* 核心：自定义Handler缓存Session，替代getSession()
*/
private WebSocketConnectionManager createWebSocketManager() {
// 1. 配置WebSocket客户端（设置超时）
StandardWebSocketClient webSocketClient = new StandardWebSocketClient();
webSocketClient.setUserProperties(Map.of(
"javax.websocket.client.connect.timeout", CONNECT_TIMEOUT_MS,
"org.apache.tomcat.websocket.connect.timeout", CONNECT_TIMEOUT_MS
));
// 2. 自定义Handler（缓存Session + 处理消息）
TextWebSocketHandler customHandler = new TextWebSocketHandler() {
// 连接建立时缓存Session
@Override
public void afterConnectionEstablished(WebSocketSession session) throws Exception {
sessionRef.set(session); // 缓存Session实例
isConnected.set(true);
log.info("WebSocket连接成功，sessionId:{}", session.getId());
super.afterConnectionEstablished(session);
}
// 处理响应消息
@Override
protected void handleTextMessage(WebSocketSession session, TextMessage message) {
try {
MDC.put(TRACE_ID, UUID.randomUUID().toString()); // 补全TRACE_ID
String payload = message.getPayload();
// 严格非空校验
UkeyResponse response = JSONUtil.toBean(payload, UkeyResponse.class, true);
if (response == null || response.get_id() == null || response.get_args() == null) {
log.error("U-Key响应格式异常，payload:{}", payload);
return;
}
// 匹配请求并完成Future
String requestId = response.get_id().toString();
CompletableFuture<UkeyResponse.Args> future = requestFutureMap.remove(requestId);
if (future == null) {
log.warn("未找到匹配的U-Key请求，id:{}", requestId);
return;
}
// 处理结果
UkeyResponse.Args args = response.get_args();
if (Boolean.FALSE.equals(args.getResult())) {
String errorMsg = Objects.requireNonNullElse(args.getError().toString(), "未知错误");
log.error("U-Key加签失败 | id={}, error={}", requestId, errorMsg);
sendErrorNotify(response, errorMsg);
future.completeExceptionally(new UkeySignException(errorMsg));
} else {
log.info("U-Key加签成功 | id={}, data={}", requestId, args.getData());
future.complete(args);
}
} catch (Exception e) {
log.error("处理U-Key响应异常", e);
failAllPendingFuture(new UkeySignException("响应处理异常", e));
} finally {
MDC.remove(TRACE_ID);
}
}
// 传输异常处理
@Override
public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
log.error("WebSocket传输异常 | sessionId={}", session.getId(), exception);
sessionRef.set(null); // 清空失效Session
isConnected.set(false);
failAllPendingFuture(new UkeyConnectionException("传输异常", exception));
super.handleTransportError(session, exception);
}
// 连接关闭处理
@Override
public void afterConnectionClosed(WebSocketSession session, org.springframework.web.socket.CloseStatus status) throws Exception {
log.warn("WebSocket连接关闭 | sessionId={}, status={}", session.getId(), status);
sessionRef.set(null); // 清空失效Session
isConnected.set(false);
failAllPendingFuture(new UkeyConnectionException("连接关闭，status:" + status));
super.afterConnectionClosed(session, status);
}
};
// 3. 创建连接管理器（仅使用通用方法，兼容所有版本）
WebSocketConnectionManager manager = new WebSocketConnectionManager(
webSocketClient, customHandler, ukeyProperties.getWsUrl()
);
manager.setAutoStartup(true); // 自动启动
return manager;
}
/**
* 获取有效WebSocketSession（替代getSession()）
*/
private WebSocketSession getValidSession() {
WebSocketSession session = sessionRef.get();
// 校验Session是否有效
if (session == null || !session.isOpen()) {
return null;
}
return session;
}
/**
* 心跳检测（保活+重连）
*/
private void startHeartbeat() {
heartbeatExecutor.scheduleAtFixedRate(() -> {
// 连接失效则重连
if (!isConnected.get()) {
reconnect();
return;
}
// 发送心跳包
WebSocketSession session = getValidSession();
if (session != null) {
try {
UkeyHeartbeatRequest heartbeat = new UkeyHeartbeatRequest();
heartbeat.setId(UUID.randomUUID().toString());
session.sendMessage(new TextMessage(JSONUtil.toJsonStr(heartbeat)));
log.debug("发送心跳包 | sessionId={}", session.getId());
} catch (Exception e) {
log.error("发送心跳包异常", e);
isConnected.set(false);
sessionRef.set(null);
}
} else {
isConnected.set(false);
}
}, 0, HEARTBEAT_INTERVAL_SEC, TimeUnit.SECONDS);
}
/**
* 重连WebSocket（加锁避免并发重连）
*/
private void reconnect() {
synchronized (connectLock) {
if (isConnected.get()) {
return;
}
log.info("开始重连WebSocket...");
try {
// 停止旧连接
if (wsConnManager.isRunning()) {
wsConnManager.stop();
}
// 启动新连接
wsConnManager.start();
// 等待连接建立（重试机制）
int retry = 0;
while (!isConnected.get() && retry < RECONNECT_MAX_RETRY) {
Thread.sleep(RECONNECT_RETRY_DELAY_MS);
retry++;
}
if (isConnected.get()) {
log.info("WebSocket重连成功（重试{}次）", retry);
} else {
log.error("WebSocket重连失败（已重试{}次）", RECONNECT_MAX_RETRY);
}
} catch (Exception e) {
log.error("WebSocket重连异常", e);
}
}
}
/**
* 失败所有待处理请求
*/
private void failAllPendingFuture(Throwable e) {
if (requestFutureMap.isEmpty()) {
return;
}
log.error("批量失败待处理请求 | 原因={}", e.getMessage());
requestFutureMap.forEach((requestId, future) -> {
future.completeExceptionally(e);
log.error("请求失败 | id={}", requestId, e);
});
requestFutureMap.clear();
}
/**
* 发送错误通知
*/
private void sendErrorNotify(UkeyResponse response, String errorMsg) {
try {
WebSocketWrapper wrapper = new WebSocketWrapper();
wrapper.setSessionId(MDC.get(TRACE_ID));
wrapper.setPayload(JSONUtil.toJsonStr(response.get_args()));
wrapper.setRequest(new SignRequest(response.get_id(), errorMsg));
wrapper.setSignResult(new SignResult().setSuccess(false));
emailNotifyStrategy.notify(new EmailNotifyRecord(wrapper, response));
} catch (Exception e) {
log.error("发送错误通知异常", e);
}
}
// -------------------- 对外核心方法 --------------------
/**
* 同步获取U-Key加签结果（兼容原有调用方式）
*/
public UkeyResponse.Args getUkeyResponseArgs(@NonNull UkeyRequest ukeyRequest) throws Exception {
// 1. 参数校验
if (ukeyRequest.get_id() == null) {
throw new IllegalArgumentException("UkeyRequest的_id不能为空");
}
String requestId = String.valueOf(ukeyRequest.get_id());
// 2. 确保连接可用
if (!isConnected.get()) {
reconnect();
if (!isConnected.get()) {
throw new UkeyConnectionException("WebSocket连接不可用");
}
}
// 3. 创建Future并关联
CompletableFuture<UkeyResponse.Args> future = new CompletableFuture<>();
requestFutureMap.put(requestId, future);
try {
// 4. 获取有效Session并发送请求
WebSocketSession session = getValidSession();
if (session == null) {
throw new UkeyConnectionException("WebSocket会话无效");
}
String requestJson = JSONUtil.toJsonStr(ukeyRequest);
log.info("发送U-Key请求 | id={}, payload={}", requestId, requestJson);
session.sendMessage(new TextMessage(requestJson));
// 5. 等待响应（带超时）
UkeyResponse.Args result = future.get(RESPONSE_TIMEOUT_MS, TimeUnit.MILLISECONDS);
if (result == null) {
throw new UkeySignException("U-Key返回空结果");
}
return result;
} catch (TimeoutException e) {
log.error("U-Key请求超时 | id={}", requestId);
requestFutureMap.remove(requestId);
throw new TimeoutException();
} catch (ExecutionException e) {
requestFutureMap.remove(requestId);
Throwable cause = e.getCause();
if (cause instanceof UkeySignException) {
throw (UkeySignException) cause;
} else if (cause instanceof UkeyConnectionException) {
throw (UkeyConnectionException) cause;
} else {
throw new UkeySignException("加签处理失败: " + cause.getMessage(), cause);
}
} catch (Exception e) {
requestFutureMap.remove(requestId);
throw new UkeyConnectionException("请求发送失败", e);
}
}
/**
* 异步获取U-Key加签结果（推荐高并发场景）
*/
public CompletableFuture<UkeyResponse.Args> getUkeyResponseArgsAsync(@NonNull UkeyRequest ukeyRequest) {
// 1. 参数校验
if (ukeyRequest.get_id() == null) {
return CompletableFuture.failedFuture(new IllegalArgumentException("UkeyRequest的_id不能为空"));
}
String requestId = ukeyRequest.get_id().toString();
// 2. 连接校验
if (!isConnected.get()) {
reconnect();
if (!isConnected.get()) {
return CompletableFuture.failedFuture(new UkeyConnectionException("WebSocket连接不可用"));
}
}
// 3. 创建Future
CompletableFuture<UkeyResponse.Args> future = new CompletableFuture<>();
requestFutureMap.put(requestId, future);
// 4. 发送请求
try {
WebSocketSession session = getValidSession();
if (session == null) {
throw new UkeyConnectionException("WebSocket会话无效");
}
String requestJson = JSONUtil.toJsonStr(ukeyRequest);
log.info("异步发送U-Key请求 | id={}, payload={}", requestId, requestJson);
session.sendMessage(new TextMessage(requestJson));
} catch (Exception e) {
requestFutureMap.remove(requestId);
return CompletableFuture.failedFuture(new UkeyConnectionException("请求发送失败", e));
}
// 5. 超时+异常处理
return future.orTimeout(RESPONSE_TIMEOUT_MS, TimeUnit.MILLISECONDS)
.exceptionally(ex -> {
requestFutureMap.remove(requestId);
log.error("异步处理失败 | id={}", requestId, ex);
if (ex instanceof TimeoutException) {
try {
throw new Exception();
} catch (Exception e) {
throw new RuntimeException(e);
}
} else if (ex instanceof UkeySignException) {
throw new CompletionException(ex);
} else {
try {
throw new Exception();
} catch (Exception e) {
throw new RuntimeException(e);
}
}
});
}
/**
* 销毁资源（Spring容器关闭时调用）
*/
@PreDestroy
public void destroy() {
// 停止心跳线程池
heartbeatExecutor.shutdown();
try {
if (!heartbeatExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
heartbeatExecutor.shutdownNow();
}
} catch (InterruptedException e) {
heartbeatExecutor.shutdownNow();
Thread.currentThread().interrupt();
}
// 停止WebSocket连接
if (wsConnManager.isRunning()) {
wsConnManager.stop();
}
// 清空缓存
sessionRef.set(null);
requestFutureMap.clear();
log.info("U-Key加签服务资源已释放");
}
// -------------------- 异常定义 --------------------
public static class UkeySignException extends Exception {
public UkeySignException(String message) {
super(message);
}
public UkeySignException(String message, Throwable cause) {
super(message, cause);
}
}
public static class UkeyConnectionException extends Exception {
public UkeyConnectionException(String message) {
super(message);
}
public UkeyConnectionException(String message, Throwable cause) {
super(message, cause);
}
}
public static class UkeyHeartbeatRequest {
private String id;
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
}
public static class FrameworkConstant {
public static final String TRACE_ID = "TRACE_ID";
}
}
chinaport-data-signature-webapp/src/test/java/cn/alphahub/eport/signature/core/v121/package-info.java
+3
View File
@@ -0,0 +1,3 @@
package cn.alphahub.eport.signature.core.v121;
// 暂时未使用该类
chinaport-data-signature-webapp/src/test/java/cn/alphahub/eport/signature/core/web/EportReportResultHttpClientTest.java
+12
View File
@@ -1,6 +1,8 @@
package cn.alphahub.eport.signature.core.web;
import cn.alphahub.eport.signature.config.ChinaEportProperties;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.DisplayName;
@@ -34,4 +36,14 @@ class EportReportResultHttpClientTest {
Assertions.assertNotNull(data, "621回执结果不为空");
});
}
@Test
@DisplayName("查询 900 报错信息")
void getCeb900msg() {
String currDate = DateTimeFormatter.ofPattern("yyyyMMddHHmmss").format(LocalDateTime.now());
eportReportResultHttpClient.getCeb900msgResult(chinaEportProperties.getDxpId(), currDate).subscribe(data -> {
log.debug("查询 900 报错信息：{}", data);
Assertions.assertNotNull(data, "311回执结果不为空");
});
}
}
chinaport-data-signature-webapp/src/test/java/cn/alphahub/eport/signature/util/CommandClientTest.java
+1 -1
View File
@@ -26,7 +26,7 @@ class CommandClientTest {
@Test
void execute2() {
CommandClient.getSharedInstance().execute("""
java -jar /Users/weasley/Development/IdeaProjects/chinaport-data-signature/target/chinaport-data-signature.jar
java -jar /Users/julian/Development/IdeaProjects/chinaport-data-signature/target/chinaport-data-signature.jar
""");
}
deploy.sh
+1 -1
View File
@@ -8,7 +8,7 @@ else
echo "未检测到 sptn 命令，跳过执行"
fi
SETTINGS="/Users/weasley/Development/program/apache-maven/conf/settings-maven-central.xml"
SETTINGS="/Users/$USER/Development/program/apache-maven/conf/settings-maven-central.xml"
MODULE="chinaport-data-signature-data-model"
clear && mvn deploy -pl :$MODULE -am --settings $SETTINGS