autumn/jsowell-charger-web
1
0
Fork 0
mirror of https://codeup.aliyun.com/67c68d4e484ca2f0a13ac3c1/ydc/jsowell-charger-web.git synced 2026-07-17 12:38:29 +08:00
Code Issues Packages Projects Releases Wiki Activity

对接jcpp

Browse Source
This commit is contained in:
Guoqs
2025-12-27 15:05:41 +08:00
parent bc3e8091de
commit 7d170dbe64
35 changed files with 5185 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

173
jsowell-pile/src/main/java/com/jsowell/pile/jcpp/service/IJcppDownlinkService.java Normal file
View File

@@ -0,0 +1,173 @@
package com.jsowell.pile.jcpp.service;
import com.jsowell.pile.jcpp.dto.JcppPricingModel;
import com.jsowell.pile.jcpp.dto.JcppSessionInfo;
import java.util.List;
import java.util.Map;
/**
* JCPP 下行调用服务接口
*
* @author jsowell
*/
public interface IJcppDownlinkService {
/**
* 发送登录应答
*
* @param pileCode 充电桩编码
* @param success 是否登录成功
*/
void sendLoginAck(String pileCode, boolean success);
/**
* 发送远程启动充电指令
*
* @param pileCode 充电桩编码
* @param gunNo 枪编号
* @param tradeNo 交易流水号
* @param limitYuan 限制金额（元）
* @param logicalCardNo 逻辑卡号
* @param physicalCardNo 物理卡号
*/
void sendRemoteStartCharging(String pileCode, String gunNo, String tradeNo,
String limitYuan, String logicalCardNo, String physicalCardNo);
/**
* 发送远程停止充电指令
*
* @param pileCode 充电桩编码
* @param gunNo 枪编号
*/
void sendRemoteStopCharging(String pileCode, String gunNo);
/**
* 发送计费模板
*
* @param pileCode 充电桩编码
* @param pricingId 计费模板ID
* @param pricingModel 计费模板
*/
void sendSetPricing(String pileCode, Long pricingId, Object pricingModel);
/**
* 发送查询计费应答
*
* @param pileCode 充电桩编码
* @param pricingId 计费模板ID
* @param pricingModel 计费模板
*/
void sendQueryPricingAck(String pileCode, Long pricingId, Object pricingModel);
/**
* 发送校验计费应答
*
* @param pileCode 充电桩编码
* @param success 是否校验成功
* @param pricingId 计费模板ID
*/
void sendVerifyPricingAck(String pileCode, boolean success, Long pricingId);
/**
* 发送启动充电应答（刷卡鉴权结果）
*
* @param pileCode 充电桩编码
* @param gunNo 枪编号
* @param tradeNo 交易流水号
* @param logicalCardNo 逻辑卡号
* @param limitYuan 限制金额（元）
* @param authSuccess 鉴权是否成功
* @param failReason 失败原因
*/
void sendStartChargeAck(String pileCode, String gunNo, String tradeNo,
String logicalCardNo, String limitYuan,
boolean authSuccess, String failReason);
/**
* 发送交易记录应答
*
* @param pileCode 充电桩编码
* @param tradeNo 交易流水号
* @param success 是否成功
*/
void sendTransactionRecordAck(String pileCode, String tradeNo, boolean success);
/**
* 检查充电桩是否在线
*
* @param pileCode 充电桩编码
* @return 是否在线
*/
boolean isPileOnline(String pileCode);
/**
* 获取充电桩会话信息
*
* @param pileCode 充电桩编码
* @return 会话信息
*/
Map<String, String> getSessionInfo(String pileCode);
// ==================== 兼容旧接口 ====================
/**
* 远程启动充电（兼容旧接口）
*/
boolean remoteStartCharging(String sessionId, String pileCode, String gunNo,
String tradeNo, String limitYuan,
String logicalCardNo, String physicalCardNo);
/**
* 远程停止充电（兼容旧接口）
*/
boolean remoteStopCharging(String sessionId, String pileCode, String gunNo);
/**
* 下发计费模板（兼容旧接口）
*/
boolean setPricing(String sessionId, String pileCode, Long pricingId, JcppPricingModel pricingModel);
/**
* 查询计费应答（兼容旧接口）
*/
boolean queryPricingAck(String sessionId, String pileCode, Long pricingId, JcppPricingModel pricingModel);
/**
* 校验计费应答（兼容旧接口）
*/
boolean verifyPricingAck(String sessionId, String pileCode, boolean success, Long pricingId);
/**
* 登录应答（兼容旧接口）
*/
boolean loginAck(String sessionId, String pileCode, boolean success);
/**
* 启动充电应答（兼容旧接口）
*/
boolean startChargeAck(String sessionId, String pileCode, String gunNo,
String tradeNo, String logicalCardNo, String limitYuan,
boolean authSuccess, String failReason);
/**
* 交易记录应答（兼容旧接口）
*/
boolean transactionRecordAck(String sessionId, String tradeNo, boolean success);
/**
* 查询充电桩会话信息（兼容旧接口）
*/
JcppSessionInfo getSession(String pileCode);
/**
* 查询所有在线充电桩（兼容旧接口）
*/
List<JcppSessionInfo> getSessions();
/**
* 发送通用下行指令（兼容旧接口）
*/
Map<String, Object> sendDownlinkCommand(String sessionId, String pileCode,
String commandType, Object data);
}

100
jsowell-pile/src/main/java/com/jsowell/pile/jcpp/service/IJcppMessageService.java Normal file
View File

@@ -0,0 +1,100 @@
package com.jsowell.pile.jcpp.service;
import com.jsowell.pile.jcpp.dto.JcppUplinkMessage;
import com.jsowell.pile.jcpp.dto.JcppUplinkResponse;
/**
* JCPP 消息处理服务接口
*
* @author jsowell
*/
public interface IJcppMessageService {
/**
* 处理上行消息
*
* @param message 上行消息
* @return 响应
*/
JcppUplinkResponse handleMessage(JcppUplinkMessage message);
/**
* 处理登录消息
*
* @param message 上行消息
* @return 响应
*/
JcppUplinkResponse handleLogin(JcppUplinkMessage message);
/**
* 处理心跳消息
*
* @param message 上行消息
* @return 响应
*/
JcppUplinkResponse handleHeartbeat(JcppUplinkMessage message);
/**
* 处理刷卡/扫码启动充电消息
*
* @param message 上行消息
* @return 响应
*/
JcppUplinkResponse handleStartCharge(JcppUplinkMessage message);
/**
* 处理实时数据上报消息
*
* @param message 上行消息
* @return 响应
*/
JcppUplinkResponse handleRealTimeData(JcppUplinkMessage message);
/**
* 处理交易记录消息
*
* @param message 上行消息
* @return 响应
*/
JcppUplinkResponse handleTransactionRecord(JcppUplinkMessage message);
/**
* 处理枪状态变化消息
*
* @param message 上行消息
* @return 响应
*/
JcppUplinkResponse handleGunStatus(JcppUplinkMessage message);
/**
* 处理校验计费模板消息
*
* @param message 上行消息
* @return 响应
*/
JcppUplinkResponse handleVerifyPricing(JcppUplinkMessage message);
/**
* 处理查询计费模板消息
*
* @param message 上行消息
* @return 响应
*/
JcppUplinkResponse handleQueryPricing(JcppUplinkMessage message);
/**
* 处理远程启动结果消息
*
* @param message 上行消息
* @return 响应
*/
JcppUplinkResponse handleRemoteStartResult(JcppUplinkMessage message);
/**
* 处理远程停止结果消息
*
* @param message 上行消息
* @return 响应
*/
JcppUplinkResponse handleRemoteStopResult(JcppUplinkMessage message);
}

46
jsowell-pile/src/main/java/com/jsowell/pile/jcpp/service/IJcppRemoteChargeService.java Normal file
View File

@@ -0,0 +1,46 @@
package com.jsowell.pile.jcpp.service;
/**
* JCPP 远程充电服务接口
* 用于 APP/小程序发起的远程启动/停止充电
*
* @author jsowell
*/
public interface IJcppRemoteChargeService {
/**
* 远程启动充电
*
* @param memberId 会员ID
* @param pileCode 充电桩编码
* @param gunNo 枪编号
* @param payAmount 预付金额（元）
* @return 订单号
*/
String remoteStartCharging(String memberId, String pileCode, String gunNo, String payAmount);
/**
* 远程停止充电
*
* @param memberId 会员ID
* @param orderCode 订单号
* @return 是否成功
*/
boolean remoteStopCharging(String memberId, String orderCode);
/**
* 检查充电桩是否在线
*
* @param pileCode 充电桩编码
* @return 是否在线
*/
boolean isPileOnline(String pileCode);
/**
* 获取充电桩会话ID
*
* @param pileCode 充电桩编码
* @return 会话ID
*/
String getSessionId(String pileCode);
}

432
jsowell-pile/src/main/java/com/jsowell/pile/jcpp/service/impl/JcppDownlinkServiceImpl.java Normal file
View File

@@ -0,0 +1,432 @@
package com.jsowell.pile.jcpp.service.impl;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.TypeReference;
import com.jsowell.pile.jcpp.config.JcppConfig;
import com.jsowell.pile.jcpp.constant.JcppConstants;
import com.jsowell.pile.jcpp.dto.JcppDownlinkCommand;
import com.jsowell.pile.jcpp.dto.JcppDownlinkRequest;
import com.jsowell.pile.jcpp.dto.JcppPricingModel;
import com.jsowell.pile.jcpp.dto.JcppSessionInfo;
import com.jsowell.pile.jcpp.service.IJcppDownlinkService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestClientException;
import org.springframework.web.client.RestTemplate;
import java.util.*;
import java.util.concurrent.TimeUnit;
/**
* JCPP 下行调用服务实现类
*
* @author jsowell
*/
@Slf4j
@Service
public class JcppDownlinkServiceImpl implements IJcppDownlinkService {
@Autowired
private JcppConfig jcppConfig;
@Autowired
@Qualifier("jcppRestTemplate")
private RestTemplate restTemplate;
@Autowired
private StringRedisTemplate stringRedisTemplate;
// ==================== 新接口实现（基于 Redis 队列） ====================
@Override
public void sendLoginAck(String pileCode, boolean success) {
Map<String, Object> data = new HashMap<>();
data.put("success", success);
sendCommand(pileCode, JcppConstants.DownlinkCommand.LOGIN_ACK, data);
}
@Override
public void sendRemoteStartCharging(String pileCode, String gunNo, String tradeNo,
String limitYuan, String logicalCardNo, String physicalCardNo) {
Map<String, Object> data = new HashMap<>();
data.put("gunNo", gunNo);
data.put("tradeNo", tradeNo);
data.put("limitYuan", limitYuan);
data.put("logicalCardNo", logicalCardNo);
data.put("physicalCardNo", physicalCardNo);
sendCommand(pileCode, JcppConstants.DownlinkCommand.REMOTE_START, data);
}
@Override
public void sendRemoteStopCharging(String pileCode, String gunNo) {
Map<String, Object> data = new HashMap<>();
data.put("gunNo", gunNo);
sendCommand(pileCode, JcppConstants.DownlinkCommand.REMOTE_STOP, data);
}
@Override
public void sendSetPricing(String pileCode, Long pricingId, Object pricingModel) {
Map<String, Object> data = new HashMap<>();
data.put("pricingId", pricingId);
data.put("pricingModel", pricingModel);
sendCommand(pileCode, JcppConstants.DownlinkCommand.SET_PRICING, data);
}
@Override
public void sendQueryPricingAck(String pileCode, Long pricingId, Object pricingModel) {
Map<String, Object> data = new HashMap<>();
data.put("pricingId", pricingId);
data.put("pricingModel", pricingModel);
sendCommand(pileCode, JcppConstants.DownlinkCommand.QUERY_PRICING_ACK, data);
}
@Override
public void sendVerifyPricingAck(String pileCode, boolean success, Long pricingId) {
Map<String, Object> data = new HashMap<>();
data.put("success", success);
data.put("pricingId", pricingId);
sendCommand(pileCode, JcppConstants.DownlinkCommand.VERIFY_PRICING_ACK, data);
}
@Override
public void sendStartChargeAck(String pileCode, String gunNo, String tradeNo,
String logicalCardNo, String limitYuan,
boolean authSuccess, String failReason) {
Map<String, Object> data = new HashMap<>();
data.put("gunNo", gunNo);
data.put("tradeNo", tradeNo);
data.put("logicalCardNo", logicalCardNo);
data.put("limitYuan", limitYuan);
data.put("authSuccess", authSuccess);
data.put("failReason", failReason);
sendCommand(pileCode, JcppConstants.DownlinkCommand.START_CHARGE_ACK, data);
}
@Override
public void sendTransactionRecordAck(String pileCode, String tradeNo, boolean success) {
Map<String, Object> data = new HashMap<>();
data.put("tradeNo", tradeNo);
data.put("success", success);
sendCommand(pileCode, JcppConstants.DownlinkCommand.TRANSACTION_RECORD_ACK, data);
}
@Override
public boolean isPileOnline(String pileCode) {
Boolean isMember = stringRedisTemplate.opsForSet().isMember(JcppConstants.REDIS_ONLINE_PILES_KEY, pileCode);
return Boolean.TRUE.equals(isMember);
}
@Override
public Map<String, String> getSessionInfo(String pileCode) {
String key = JcppConstants.REDIS_SESSION_PREFIX + pileCode;
Map<Object, Object> entries = stringRedisTemplate.opsForHash().entries(key);
Map<String, String> result = new HashMap<>();
entries.forEach((k, v) -> result.put(String.valueOf(k), String.valueOf(v)));
return result;
}
/**
* 发送下行指令到 Redis 队列
*/
private void sendCommand(String pileCode, String commandType, Object data) {
JcppDownlinkCommand command = JcppDownlinkCommand.builder()
.commandId(UUID.randomUUID().toString())
.commandType(commandType)
.pileCode(pileCode)
.timestamp(System.currentTimeMillis())
.data(data)
.build();
String key = JcppConstants.REDIS_DOWNLINK_PREFIX + pileCode;
String json = JSON.toJSONString(command);
stringRedisTemplate.opsForList().rightPush(key, json);
log.info("发送 JCPP 下行指令: pileCode={}, commandType={}, commandId={}",
pileCode, commandType, command.getCommandId());
}
// ==================== 兼容旧接口实现（基于 HTTP） ====================
@Override
public boolean remoteStartCharging(String sessionId, String pileCode, String gunNo,
String tradeNo, String limitYuan,
String logicalCardNo, String physicalCardNo) {
Map<String, Object> data = new HashMap<>();
data.put("gunNo", gunNo);
data.put("tradeNo", tradeNo);
data.put("limitYuan", limitYuan);
data.put("logicalCardNo", logicalCardNo);
data.put("physicalCardNo", physicalCardNo);
Map<String, Object> result = sendDownlinkCommand(sessionId, pileCode,
JcppConstants.DownlinkCommand.REMOTE_START, data);
return isSuccess(result);
}
@Override
public boolean remoteStopCharging(String sessionId, String pileCode, String gunNo) {
Map<String, Object> data = new HashMap<>();
data.put("gunNo", gunNo);
Map<String, Object> result = sendDownlinkCommand(sessionId, pileCode,
JcppConstants.DownlinkCommand.REMOTE_STOP, data);
return isSuccess(result);
}
@Override
public boolean setPricing(String sessionId, String pileCode, Long pricingId, JcppPricingModel pricingModel) {
Map<String, Object> data = new HashMap<>();
data.put("pricingId", pricingId);
data.put("pricingModel", pricingModel);
Map<String, Object> result = sendDownlinkCommand(sessionId, pileCode,
JcppConstants.DownlinkCommand.SET_PRICING, data);
return isSuccess(result);
}
@Override
public boolean queryPricingAck(String sessionId, String pileCode, Long pricingId, JcppPricingModel pricingModel) {
Map<String, Object> data = new HashMap<>();
data.put("pricingId", pricingId);
data.put("pricingModel", pricingModel);
Map<String, Object> result = sendDownlinkCommand(sessionId, pileCode,
JcppConstants.DownlinkCommand.QUERY_PRICING_ACK, data);
return isSuccess(result);
}
@Override
public boolean verifyPricingAck(String sessionId, String pileCode, boolean success, Long pricingId) {
Map<String, Object> data = new HashMap<>();
data.put("success", success);
data.put("pricingId", pricingId);
Map<String, Object> result = sendDownlinkCommand(sessionId, pileCode,
JcppConstants.DownlinkCommand.VERIFY_PRICING_ACK, data);
return isSuccess(result);
}
@Override
public boolean loginAck(String sessionId, String pileCode, boolean success) {
Map<String, Object> data = new HashMap<>();
data.put("success", success);
Map<String, Object> result = sendDownlinkCommand(sessionId, pileCode,
JcppConstants.DownlinkCommand.LOGIN_ACK, data);
return isSuccess(result);
}
@Override
public boolean startChargeAck(String sessionId, String pileCode, String gunNo,
String tradeNo, String logicalCardNo, String limitYuan,
boolean authSuccess, String failReason) {
Map<String, Object> data = new HashMap<>();
data.put("gunNo", gunNo);
data.put("tradeNo", tradeNo);
data.put("logicalCardNo", logicalCardNo);
data.put("limitYuan", limitYuan);
data.put("authSuccess", authSuccess);
data.put("failReason", failReason);
Map<String, Object> result = sendDownlinkCommand(sessionId, pileCode,
JcppConstants.DownlinkCommand.START_CHARGE_ACK, data);
return isSuccess(result);
}
@Override
public boolean transactionRecordAck(String sessionId, String tradeNo, boolean success) {
Map<String, Object> data = new HashMap<>();
data.put("tradeNo", tradeNo);
data.put("success", success);
// 从 Redis 获取 pileCode
String pileCode = getPileCodeByTradeNo(tradeNo);
if (pileCode == null) {
log.warn("无法获取交易流水号对应的充电桩编码, tradeNo: {}", tradeNo);
return false;
}
Map<String, Object> result = sendDownlinkCommand(sessionId, pileCode,
JcppConstants.DownlinkCommand.TRANSACTION_RECORD_ACK, data);
return isSuccess(result);
}
@Override
public JcppSessionInfo getSession(String pileCode) {
// 先从 Redis 缓存获取
String cacheKey = JcppConstants.REDIS_KEY_SESSION + pileCode;
String sessionJson = stringRedisTemplate.opsForValue().get(cacheKey);
if (sessionJson != null) {
return JSON.parseObject(sessionJson, JcppSessionInfo.class);
}
// 从 JCPP 服务获取
try {
String url = jcppConfig.getSessionUrl() + "/" + pileCode;
ResponseEntity<String> response = restTemplate.getForEntity(url, String.class);
if (response.getStatusCode().is2xxSuccessful() && response.getBody() != null) {
Map<String, Object> result = JSON.parseObject(response.getBody(),
new TypeReference<Map<String, Object>>() {});
if (Boolean.TRUE.equals(result.get("success"))) {
Object data = result.get("data");
if (data != null) {
JcppSessionInfo sessionInfo = JSON.parseObject(JSON.toJSONString(data), JcppSessionInfo.class);
// 缓存到 Redis
stringRedisTemplate.opsForValue().set(cacheKey, JSON.toJSONString(sessionInfo),
JcppConstants.SESSION_EXPIRE_SECONDS, TimeUnit.SECONDS);
return sessionInfo;
}
}
}
} catch (RestClientException e) {
log.error("获取 JCPP 会话信息失败, pileCode: {}, error: {}", pileCode, e.getMessage());
}
return null;
}
@Override
public List<JcppSessionInfo> getSessions() {
try {
String url = jcppConfig.getSessionUrl();
ResponseEntity<String> response = restTemplate.getForEntity(url, String.class);
if (response.getStatusCode().is2xxSuccessful() && response.getBody() != null) {
Map<String, Object> result = JSON.parseObject(response.getBody(),
new TypeReference<Map<String, Object>>() {});
if (Boolean.TRUE.equals(result.get("success"))) {
Object data = result.get("data");
if (data != null) {
return JSON.parseArray(JSON.toJSONString(data), JcppSessionInfo.class);
}
}
}
} catch (RestClientException e) {
log.error("获取 JCPP 所有会话信息失败, error: {}", e.getMessage());
}
return Collections.emptyList();
}
@Override
public Map<String, Object> sendDownlinkCommand(String sessionId, String pileCode,
String commandType, Object data) {
if (!jcppConfig.isEnabled()) {
log.warn("JCPP 对接未启用，忽略下行指令: commandType={}, pileCode={}", commandType, pileCode);
return createErrorResult("JCPP 对接未启用");
}
// 如果没有传入 sessionId，尝试从 Redis 获取
if (sessionId == null || sessionId.isEmpty()) {
sessionId = getSessionIdFromRedis(pileCode);
if (sessionId == null) {
log.warn("无法获取充电桩会话ID, pileCode: {}", pileCode);
return createErrorResult("充电桩不在线");
}
}
JcppDownlinkRequest request = JcppDownlinkRequest.builder()
.sessionId(sessionId)
.pileCode(pileCode)
.commandType(commandType)
.data(data)
.requestId(UUID.randomUUID().toString())
.timestamp(System.currentTimeMillis())
.build();
return sendWithRetry(request);
}
/**
* 带重试的发送
*/
private Map<String, Object> sendWithRetry(JcppDownlinkRequest request) {
int retryCount = jcppConfig.getRetryCount();
int retryInterval = jcppConfig.getRetryInterval();
for (int i = 0; i <= retryCount; i++) {
try {
return doSend(request);
} catch (RestClientException e) {
log.warn("发送 JCPP 下行指令失败, 第 {} 次尝试, requestId: {}, error: {}",
i + 1, request.getRequestId(), e.getMessage());
if (i < retryCount) {
try {
Thread.sleep(retryInterval);
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
break;
}
}
}
}
return createErrorResult("发送下行指令失败，已重试 " + retryCount + " 次");
}
/**
* 实际发送
*/
private Map<String, Object> doSend(JcppDownlinkRequest request) {
String url = jcppConfig.getDownlinkUrl();
HttpHeaders headers = new HttpHeaders();
headers.setContentType(MediaType.APPLICATION_JSON);
HttpEntity<JcppDownlinkRequest> entity = new HttpEntity<>(request, headers);
log.info("发送 JCPP 下行指令: url=, requestId={}, commandType={}, pileCode={}",
url, request.getRequestId(), request.getCommandType(), request.getPileCode());
ResponseEntity<String> response = restTemplate.postForEntity(url, entity, String.class);
if (response.getStatusCode().is2xxSuccessful() && response.getBody() != null) {
Map<String, Object> result = JSON.parseObject(response.getBody(),
new TypeReference<Map<String, Object>>() {});
log.info("JCPP 下行指令响应: requestId={}, result={}", request.getRequestId(), result);
return result;
}
return createErrorResult("HTTP 响应异常: " + response.getStatusCode());
}
/**
* 从 Redis 获取会话ID
*/
private String getSessionIdFromRedis(String pileCode) {
String cacheKey = JcppConstants.REDIS_KEY_SESSION + pileCode;
String sessionJson = stringRedisTemplate.opsForValue().get(cacheKey);
if (sessionJson != null) {
JcppSessionInfo sessionInfo = JSON.parseObject(sessionJson, JcppSessionInfo.class);
return sessionInfo.getSessionId();
}
return null;
}
/**
* 根据交易流水号获取充电桩编码
*/
private String getPileCodeByTradeNo(String tradeNo) {
// TODO: 从订单表或 Redis 中获取
return null;
}
/**
* 判断结果是否成功
*/
private boolean isSuccess(Map<String, Object> result) {
return result != null && Boolean.TRUE.equals(result.get("success"));
}
/**
* 创建错误结果
*/
private Map<String, Object> createErrorResult(String message) {
Map<String, Object> result = new HashMap<>();
result.put("success", false);
result.put("message", message);
return result;
}
}

627
jsowell-pile/src/main/java/com/jsowell/pile/jcpp/service/impl/JcppMessageServiceImpl.java Normal file
View File

@@ -0,0 +1,627 @@
package com.jsowell.pile.jcpp.service.impl;
import com.alibaba.fastjson2.JSON;
import com.jsowell.common.util.id.IdUtils;
import com.jsowell.pile.domain.*;
import com.jsowell.pile.jcpp.constant.JcppConstants;
import com.jsowell.pile.jcpp.dto.*;
import com.jsowell.pile.jcpp.service.IJcppDownlinkService;
import com.jsowell.pile.jcpp.service.IJcppMessageService;
import com.jsowell.pile.jcpp.util.PricingModelConverter;
import com.jsowell.pile.service.*;
import com.jsowell.pile.vo.web.BillingTemplateVO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.math.BigDecimal;
import java.util.Date;
import java.util.concurrent.TimeUnit;
/**
* JCPP 消息处理服务实现类
*
* @author jsowell
*/
@Slf4j
@Service
public class JcppMessageServiceImpl implements IJcppMessageService {
@Autowired
private PileBasicInfoService pileBasicInfoService;
@Autowired
private PileBillingTemplateService pileBillingTemplateService;
@Autowired
private PileAuthCardService pileAuthCardService;
@Autowired
private MemberBasicInfoService memberBasicInfoService;
@Autowired
private MemberWalletInfoService memberWalletInfoService;
@Autowired
private OrderBasicInfoService orderBasicInfoService;
@Autowired
private PileConnectorInfoService pileConnectorInfoService;
@Autowired
private IJcppDownlinkService jcppDownlinkService;
@Autowired
private StringRedisTemplate stringRedisTemplate;
@Override
public JcppUplinkResponse handleMessage(JcppUplinkMessage message) {
if (message == null || message.getMessageType() == null) {
log.warn("收到无效的 JCPP 消息: {}", message);
return JcppUplinkResponse.error("无效的消息");
}
String messageType = message.getMessageType();
log.info("收到 JCPP 上行消息, messageId: {}, messageType: {}, pileCode: {}",
message.getMessageId(), messageType, message.getPileCode());
try {
switch (messageType) {
case JcppConstants.MessageType.LOGIN:
return handleLogin(message);
case JcppConstants.MessageType.HEARTBEAT:
return handleHeartbeat(message);
case JcppConstants.MessageType.START_CHARGE:
return handleStartCharge(message);
case JcppConstants.MessageType.REAL_TIME_DATA:
return handleRealTimeData(message);
case JcppConstants.MessageType.TRANSACTION_RECORD:
return handleTransactionRecord(message);
case JcppConstants.MessageType.GUN_STATUS:
return handleGunStatus(message);
case JcppConstants.MessageType.VERIFY_PRICING:
return handleVerifyPricing(message);
case JcppConstants.MessageType.QUERY_PRICING:
return handleQueryPricing(message);
case JcppConstants.MessageType.REMOTE_START_RESULT:
return handleRemoteStartResult(message);
case JcppConstants.MessageType.REMOTE_STOP_RESULT:
return handleRemoteStopResult(message);
default:
log.warn("未知的消息类型: {}", messageType);
return JcppUplinkResponse.error(message.getMessageId(), "未知的消息类型: " + messageType);
}
} catch (Exception e) {
log.error("处理 JCPP 消息异常, messageId: {}, messageType: {}, error: {}",
message.getMessageId(), messageType, e.getMessage(), e);
return JcppUplinkResponse.error(message.getMessageId(), "处理消息异常: " + e.getMessage());
}
}
@Override
@Transactional(rollbackFor = Exception.class)
public JcppUplinkResponse handleLogin(JcppUplinkMessage message) {
String pileCode = message.getPileCode();
String sessionId = message.getSessionId();
log.info("处理登录消息, pileCode: {}, sessionId: {}", pileCode, sessionId);
// 解析登录数据
JcppLoginData loginData = parseData(message.getData(), JcppLoginData.class);
if (loginData == null) {
log.warn("登录消息数据解析失败, pileCode: {}", pileCode);
jcppDownlinkService.loginAck(sessionId, pileCode, false);
return JcppUplinkResponse.error(message.getMessageId(), "登录数据解析失败");
}
// 查询充电桩是否存在
PileBasicInfo pileInfo = pileBasicInfoService.selectPileBasicInfoBySN(pileCode);
if (pileInfo == null) {
log.warn("充电桩不存在, pileCode: {}", pileCode);
jcppDownlinkService.loginAck(sessionId, pileCode, false);
return JcppUplinkResponse.error(message.getMessageId(), "充电桩不存在");
}
// 检查充电桩是否被删除
if ("1".equals(pileInfo.getDelFlag())) {
log.warn("充电桩已被删除, pileCode: {}", pileCode);
jcppDownlinkService.loginAck(sessionId, pileCode, false);
return JcppUplinkResponse.error(message.getMessageId(), "充电桩已被删除");
}
// 保存会话信息到 Redis
JcppSessionInfo sessionInfo = JcppSessionInfo.builder()
.sessionId(sessionId)
.pileCode(pileCode)
.remoteAddress(loginData.getRemoteAddress())
.nodeId(loginData.getNodeId())
.nodeHostAddress(loginData.getNodeHostAddress())
.nodeRestPort(loginData.getNodeRestPort())
.nodeGrpcPort(loginData.getNodeGrpcPort())
.protocolName(message.getProtocolName())
.loginTimestamp(System.currentTimeMillis())
.lastActiveTimestamp(System.currentTimeMillis())
.online(true)
.build();
// 保存会话到 Redis
String sessionKey = JcppConstants.REDIS_KEY_SESSION + pileCode;
stringRedisTemplate.opsForValue().set(sessionKey, JSON.toJSONString(sessionInfo),
JcppConstants.SESSION_EXPIRE_SECONDS, TimeUnit.SECONDS);
// 保存节点信息到 Redis
String nodeKey = JcppConstants.REDIS_KEY_NODE + pileCode;
stringRedisTemplate.opsForValue().set(nodeKey, JSON.toJSONString(loginData),
JcppConstants.SESSION_EXPIRE_SECONDS, TimeUnit.SECONDS);
// 设置在线状态
String onlineKey = JcppConstants.REDIS_KEY_ONLINE + pileCode;
stringRedisTemplate.opsForValue().set(onlineKey, "1",
JcppConstants.ONLINE_EXPIRE_SECONDS, TimeUnit.SECONDS);
log.info("充电桩登录成功, pileCode: {}, sessionId: {}, remoteAddress: {}",
pileCode, sessionId, loginData.getRemoteAddress());
// 发送登录应答
jcppDownlinkService.loginAck(sessionId, pileCode, true);
return JcppUplinkResponse.success(message.getMessageId(), null);
}
@Override
public JcppUplinkResponse handleHeartbeat(JcppUplinkMessage message) {
String pileCode = message.getPileCode();
String sessionId = message.getSessionId();
log.debug("处理心跳消息, pileCode: {}, sessionId: {}", pileCode, sessionId);
// 更新会话过期时间
String sessionKey = JcppConstants.REDIS_KEY_SESSION + pileCode;
String sessionJson = stringRedisTemplate.opsForValue().get(sessionKey);
if (sessionJson != null) {
JcppSessionInfo sessionInfo = JSON.parseObject(sessionJson, JcppSessionInfo.class);
sessionInfo.setLastActiveTimestamp(System.currentTimeMillis());
stringRedisTemplate.opsForValue().set(sessionKey, JSON.toJSONString(sessionInfo),
JcppConstants.SESSION_EXPIRE_SECONDS, TimeUnit.SECONDS);
}
// 更新节点信息过期时间
String nodeKey = JcppConstants.REDIS_KEY_NODE + pileCode;
stringRedisTemplate.expire(nodeKey, JcppConstants.SESSION_EXPIRE_SECONDS, TimeUnit.SECONDS);
// 更新在线状态过期时间
String onlineKey = JcppConstants.REDIS_KEY_ONLINE + pileCode;
stringRedisTemplate.opsForValue().set(onlineKey, "1",
JcppConstants.ONLINE_EXPIRE_SECONDS, TimeUnit.SECONDS);
return JcppUplinkResponse.success(message.getMessageId(), null);
}
@Override
@Transactional(rollbackFor = Exception.class)
public JcppUplinkResponse handleStartCharge(JcppUplinkMessage message) {
String pileCode = message.getPileCode();
String sessionId = message.getSessionId();
log.info("处理启动充电消息, pileCode: {}, sessionId: {}", pileCode, sessionId);
// 解析启动充电数据
JcppStartChargeData startData = parseData(message.getData(), JcppStartChargeData.class);
if (startData == null) {
log.warn("启动充电数据解析失败, pileCode: {}", pileCode);
return JcppUplinkResponse.error(message.getMessageId(), "启动充电数据解析失败");
}
String gunNo = startData.getGunNo();
String cardNo = startData.getCardNo();
String startType = startData.getStartType();
try {
// 查询充电桩信息
PileBasicInfo pileInfo = pileBasicInfoService.selectPileBasicInfoBySN(pileCode);
if (pileInfo == null) {
sendStartChargeAck(sessionId, pileCode, gunNo, null, cardNo, null,
false, JcppConstants.AuthFailReason.PILE_DISABLED);
return JcppUplinkResponse.error(message.getMessageId(), "充电桩不存在");
}
// 根据启动类型进行鉴权
String memberId = null;
String merchantId = String.valueOf(pileInfo.getMerchantId());
BigDecimal balance = BigDecimal.ZERO;
if (JcppConstants.StartType.CARD.equals(startType)) {
// 刷卡启动 - 根据逻辑卡号查询授权卡
PileAuthCard authCard = pileAuthCardService.selectCardInfoByLogicCard(cardNo);
if (authCard == null) {
log.warn("授权卡不存在, cardNo: {}", cardNo);
sendStartChargeAck(sessionId, pileCode, gunNo, null, cardNo, null,
false, JcppConstants.AuthFailReason.ACCOUNT_NOT_EXISTS);
return JcppUplinkResponse.error(message.getMessageId(), "授权卡不存在");
}
// 检查卡状态
if (!"1".equals(authCard.getStatus())) {
log.warn("授权卡已停用, cardNo: {}", cardNo);
sendStartChargeAck(sessionId, pileCode, gunNo, null, cardNo, null,
false, JcppConstants.AuthFailReason.ACCOUNT_FROZEN);
return JcppUplinkResponse.error(message.getMessageId(), "授权卡已停用");
}
memberId = authCard.getMemberId();
} else if (JcppConstants.StartType.VIN.equals(startType)) {
// VIN码启动
String vinCode = startData.getCarVinCode();
// TODO: 根据VIN码查询会员信息
log.info("VIN码启动, vinCode: ", vinCode);
}
// 查询会员信息和余额
if (memberId != null) {
MemberBasicInfo memberInfo = memberBasicInfoService.selectInfoByMemberId(memberId);
if (memberInfo == null) {
sendStartChargeAck(sessionId, pileCode, gunNo, null, cardNo, null,
false, JcppConstants.AuthFailReason.ACCOUNT_NOT_EXISTS);
return JcppUplinkResponse.error(message.getMessageId(), "会员不存在");
}
// 检查会员状态
if ("1".equals(memberInfo.getStatus())) {
sendStartChargeAck(sessionId, pileCode, gunNo, null, cardNo, null,
false, JcppConstants.AuthFailReason.ACCOUNT_FROZEN);
return JcppUplinkResponse.error(message.getMessageId(), "会员账户已冻结");
}
// 查询钱包余额
MemberWalletInfo walletInfo = memberWalletInfoService.selectByMemberId(memberId, merchantId);
if (walletInfo != null) {
BigDecimal principalBalance = walletInfo.getPrincipalBalance() != null ? walletInfo.getPrincipalBalance() : BigDecimal.ZERO;
BigDecimal giftBalance = walletInfo.getGiftBalance() != null ? walletInfo.getGiftBalance() : BigDecimal.ZERO;
balance = principalBalance.add(giftBalance);
}
// 检查余额是否充足（最低1元）
if (balance.compareTo(BigDecimal.ONE) < 0) {
log.warn("余额不足, memberId: {}, balance: {}", memberId, balance);
sendStartChargeAck(sessionId, pileCode, gunNo, null, cardNo, null,
false, JcppConstants.AuthFailReason.INSUFFICIENT_BALANCE);
return JcppUplinkResponse.error(message.getMessageId(), "余额不足");
}
}
// 生成交易流水号
String tradeNo = IdUtils.fastSimpleUUID();
// 创建订单
OrderBasicInfo order = new OrderBasicInfo();
order.setOrderCode(tradeNo);
order.setTransactionCode(tradeNo);
order.setPileSn(pileCode);
order.setConnectorCode(pileCode + "-" + gunNo);
order.setMemberId(memberId);
order.setMerchantId(String.valueOf(pileInfo.getMerchantId()));
order.setStationId(String.valueOf(pileInfo.getStationId()));
order.setOrderStatus("1"); // 充电中
order.setPayStatus("0"); // 待支付
order.setChargeStartTime(new Date());
order.setCreateTime(new Date());
orderBasicInfoService.insert(order);
// 保存交易流水号与充电桩的映射关系到 Redis
String tradeKey = "jcpp:trade:" + tradeNo;
stringRedisTemplate.opsForValue().set(tradeKey, pileCode, 24, TimeUnit.HOURS);
// 发送启动充电应答
sendStartChargeAck(sessionId, pileCode, gunNo, tradeNo, cardNo, balance.toString(), true, null);
log.info("刷卡启动充电鉴权成功, pileCode: {}, gunNo: {}, tradeNo: {}, memberId: {}, balance: {}",
pileCode, gunNo, tradeNo, memberId, balance);
return JcppUplinkResponse.success(message.getMessageId(), tradeNo);
} catch (Exception e) {
log.error("处理启动充电消息异常, pileCode: {}, error: {}", pileCode, e.getMessage(), e);
sendStartChargeAck(sessionId, pileCode, gunNo, null, cardNo, null,
false, JcppConstants.AuthFailReason.SYSTEM_ERROR);
return JcppUplinkResponse.error(message.getMessageId(), "处理启动充电异常: " + e.getMessage());
}
}
/**
* 发送启动充电应答
*/
private void sendStartChargeAck(String sessionId, String pileCode, String gunNo,
String tradeNo, String cardNo, String limitYuan,
boolean success, String failReason) {
try {
jcppDownlinkService.startChargeAck(sessionId, pileCode, gunNo, tradeNo, cardNo, limitYuan, success, failReason);
} catch (Exception e) {
log.error("发送启动充电应答失败, pileCode: {}, error: {}", pileCode, e.getMessage());
}
}
@Override
public JcppUplinkResponse handleRealTimeData(JcppUplinkMessage message) {
String pileCode = message.getPileCode();
log.debug("处理实时数据消息, pileCode: {}, sessionId: {}", pileCode, message.getSessionId());
// 解析实时数据
JcppRealTimeData realTimeData = parseData(message.getData(), JcppRealTimeData.class);
if (realTimeData == null) {
log.warn("实时数据解析失败, pileCode: {}", pileCode);
return JcppUplinkResponse.error(message.getMessageId(), "实时数据解析失败");
}
String tradeNo = realTimeData.getTradeNo();
String gunNo = realTimeData.getGunNo();
try {
// 根据交易流水号查询订单
OrderBasicInfo order = orderBasicInfoService.getOrderInfoByTransactionCode(tradeNo);
if (order == null) {
log.warn("订单不存在, tradeNo: {}", tradeNo);
return JcppUplinkResponse.error(message.getMessageId(), "订单不存在");
}
// 更新订单实时数据
OrderBasicInfo updateOrder = new OrderBasicInfo();
updateOrder.setId(order.getId());
if (realTimeData.getTotalChargingEnergyKWh() != null) {
// updateOrder.setTotalElectricity(new BigDecimal(realTimeData.getTotalChargingEnergyKWh()));
}
if (realTimeData.getTotalChargingCostYuan() != null) {
// updateOrder.setTotalAmount(new BigDecimal(realTimeData.getTotalChargingCostYuan()));
}
if (realTimeData.getTotalChargingDurationMin() != null) {
// updateOrder.setTotalTime(realTimeData.getTotalChargingDurationMin());
}
updateOrder.setUpdateTime(new Date());
orderBasicInfoService.updateByPrimaryKeySelective(updateOrder);
// 保存实时数据到 Redis（用于实时查询）
String realtimeKey = "jcpp:realtime:" + tradeNo;
stringRedisTemplate.opsForValue().set(realtimeKey, JSON.toJSONString(realTimeData), 10, TimeUnit.MINUTES);
// 保存监控数据到数据库（可选，根据需要控制写入频率）
// saveMonitorData(order, realTimeData);
log.debug("实时数据处理成功, tradeNo: {}, energy: {}kWh, cost: {}元",
tradeNo, realTimeData.getTotalChargingEnergyKWh(), realTimeData.getTotalChargingCostYuan());
return JcppUplinkResponse.success(message.getMessageId(), null);
} catch (Exception e) {
log.error("处理实时数据异常, tradeNo: {}, error: {}", tradeNo, e.getMessage(), e);
return JcppUplinkResponse.error(message.getMessageId(), "处理实时数据异常: " + e.getMessage());
}
}
@Override
@Transactional(rollbackFor = Exception.class)
public JcppUplinkResponse handleTransactionRecord(JcppUplinkMessage message) {
String pileCode = message.getPileCode();
String sessionId = message.getSessionId();
log.info("处理交易记录消息, pileCode: {}, sessionId: {}", pileCode, sessionId);
// 解析交易记录数据
JcppTransactionData transactionData = parseData(message.getData(), JcppTransactionData.class);
if (transactionData == null) {
log.warn("交易记录数据解析失败, pileCode: {}", pileCode);
return JcppUplinkResponse.error(message.getMessageId(), "交易记录数据解析失败");
}
String tradeNo = transactionData.getTradeNo();
try {
// 根据交易流水号查询订单
OrderBasicInfo order = orderBasicInfoService.getOrderInfoByTransactionCode(tradeNo);
if (order == null) {
log.warn("订单不存在, tradeNo: {}", tradeNo);
// 仍然发送确认，避免充电桩重复上报
jcppDownlinkService.transactionRecordAck(sessionId, tradeNo, true);
return JcppUplinkResponse.error(message.getMessageId(), "订单不存在");
}
// 幂等性检查：如果订单已经是完成状态，直接返回成功
if ("2".equals(order.getOrderStatus())) {
log.info("订单已完成，忽略重复的交易记录, tradeNo: {}", tradeNo);
jcppDownlinkService.transactionRecordAck(sessionId, tradeNo, true);
return JcppUplinkResponse.success(message.getMessageId(), null);
}
// 更新订单信息
OrderBasicInfo updateOrder = new OrderBasicInfo();
updateOrder.setId(order.getId());
updateOrder.setOrderStatus("2"); // 充电完成
updateOrder.setChargeEndTime(transactionData.getEndTs() != null ? new Date(transactionData.getEndTs()) : new Date());
updateOrder.setReason(transactionData.getStopReason());
// 设置电量
if (transactionData.getTotalEnergyKWh() != null) {
// updateOrder.setTotalElectricity(new BigDecimal(transactionData.getTotalEnergyKWh()));
}
// 设置金额（如果充电桩上报了金额则使用，否则需要根据计费模板计算）
if (transactionData.getTotalAmountYuan() != null) {
// updateOrder.setTotalAmount(new BigDecimal(transactionData.getTotalAmountYuan()));
}
// 计算充电时长
if (transactionData.getStartTs() != null && transactionData.getEndTs() != null) {
long durationMin = (transactionData.getEndTs() - transactionData.getStartTs()) / 60000;
// updateOrder.setTotalTime((int) durationMin);
}
updateOrder.setUpdateTime(new Date());
orderBasicInfoService.updateByPrimaryKeySelective(updateOrder);
// 更新枪状态为空闲
String connectorCode = pileCode + "-" + transactionData.getGunNo();
// pileConnectorInfoService.updateConnectorStatus(connectorCode, "0"); // 空闲
// 发送交易记录确认
jcppDownlinkService.transactionRecordAck(sessionId, tradeNo, true);
// 清理 Redis 中的实时数据
String realtimeKey = "jcpp:realtime:" + tradeNo;
stringRedisTemplate.delete(realtimeKey);
log.info("交易记录处理成功, tradeNo: {}, energy: {}kWh, amount: {}元, stopReason: {}",
tradeNo, transactionData.getTotalEnergyKWh(),
transactionData.getTotalAmountYuan(), transactionData.getStopReason());
// TODO: 触发结算流程（如果是预付费模式）
// orderBasicInfoService.realTimeOrderSplit(order);
return JcppUplinkResponse.success(message.getMessageId(), null);
} catch (Exception e) {
log.error("处理交易记录异常, tradeNo: {}, error: {}", tradeNo, e.getMessage(), e);
return JcppUplinkResponse.error(message.getMessageId(), "处理交易记录异常: " + e.getMessage());
}
}
@Override
public JcppUplinkResponse handleGunStatus(JcppUplinkMessage message) {
String pileCode = message.getPileCode();
log.info("处理枪状态消息, pileCode: {}, sessionId: {}", pileCode, message.getSessionId());
// TODO: 解析枪状态数据并更新数据库
// 枪状态变化通常用于更新充电枪的实时状态
return JcppUplinkResponse.success(message.getMessageId(), null);
}
@Override
public JcppUplinkResponse handleVerifyPricing(JcppUplinkMessage message) {
String pileCode = message.getPileCode();
String sessionId = message.getSessionId();
log.info("处理校验计费消息, pileCode: {}, sessionId: {}", pileCode, sessionId);
// 校验计费模板：充电桩上报当前使用的计费模板ID，平台校验是否一致
// 如果不一致，需要重新下发计费模板
try {
// 获取平台当前的计费模板
BillingTemplateVO billingTemplate = pileBillingTemplateService.selectBillingTemplateDetailByPileSn(pileCode);
Long currentPricingId = billingTemplate != null ? Long.parseLong(billingTemplate.getTemplateId()) : null;
// TODO: 从消息中获取充电桩上报的计费模板ID进行比对
// 这里简化处理，直接返回成功
jcppDownlinkService.verifyPricingAck(sessionId, pileCode, true, currentPricingId);
return JcppUplinkResponse.success(message.getMessageId(), null);
} catch (Exception e) {
log.error("校验计费模板失败, pileCode: , error: {}", pileCode, e.getMessage(), e);
jcppDownlinkService.verifyPricingAck(sessionId, pileCode, false, null);
return JcppUplinkResponse.error(message.getMessageId(), "校验计费模板失败: " + e.getMessage());
}
}
@Override
public JcppUplinkResponse handleQueryPricing(JcppUplinkMessage message) {
String pileCode = message.getPileCode();
String sessionId = message.getSessionId();
log.info("处理查询计费消息, pileCode: {}, sessionId: {}", pileCode, sessionId);
try {
// 根据充电桩编码查询计费模板
BillingTemplateVO billingTemplate = pileBillingTemplateService.selectBillingTemplateDetailByPileSn(pileCode);
if (billingTemplate == null) {
log.warn("未找到充电桩的计费模板, pileCode: {}", pileCode);
jcppDownlinkService.queryPricingAck(sessionId, pileCode, null, null);
return JcppUplinkResponse.error(message.getMessageId(), "未找到计费模板");
}
// 转换为 JCPP 计费模板格式
JcppPricingModel pricingModel = PricingModelConverter.convert(billingTemplate);
// 发送计费模板应答
jcppDownlinkService.queryPricingAck(sessionId, pileCode, Long.parseLong(billingTemplate.getTemplateId()), pricingModel);
log.info("查询计费模板成功, pileCode: {}, templateId: {}, templateName: {}",
pileCode, billingTemplate.getTemplateId(), billingTemplate.getTemplateName());
return JcppUplinkResponse.success(message.getMessageId(), pricingModel);
} catch (Exception e) {
log.error("查询计费模板失败, pileCode: {}, error: {}", pileCode, e.getMessage(), e);
return JcppUplinkResponse.error(message.getMessageId(), "查询计费模板失败: " + e.getMessage());
}
}
@Override
@Transactional(rollbackFor = Exception.class)
public JcppUplinkResponse handleRemoteStartResult(JcppUplinkMessage message) {
String pileCode = message.getPileCode();
String sessionId = message.getSessionId();
log.info("处理远程启动结果消息, pileCode: {}, sessionId: {}", pileCode, sessionId);
// 解析远程启动结果数据
JcppRemoteStartResultData resultData = parseData(message.getData(), JcppRemoteStartResultData.class);
if (resultData == null) {
log.warn("远程启动结果数据解析失败, pileCode: {}", pileCode);
return JcppUplinkResponse.error(message.getMessageId(), "远程启动结果数据解析失败");
}
String tradeNo = resultData.getTradeNo();
boolean success = Boolean.TRUE.equals(resultData.getSuccess());
try {
// 根据交易流水号查询订单
OrderBasicInfo order = orderBasicInfoService.getOrderInfoByTransactionCode(tradeNo);
if (order == null) {
log.warn("订单不存在, tradeNo: {}", tradeNo);
return JcppUplinkResponse.error(message.getMessageId(), "订单不存在");
}
OrderBasicInfo updateOrder = new OrderBasicInfo();
updateOrder.setId(order.getId());
updateOrder.setUpdateTime(new Date());
if (success) {
// 启动成功
updateOrder.setOrderStatus("1"); // 充电中
log.info("远程启动充电成功, tradeNo: {}, pileCode: {}, gunNo: {}",
tradeNo, pileCode, resultData.getGunNo());
} else {
// 启动失败
updateOrder.setOrderStatus("4"); // 异常结束
updateOrder.setReason(resultData.getFailReason());
log.warn("远程启动充电失败, tradeNo: {}, pileCode: {}, failReason: {}",
tradeNo, pileCode, resultData.getFailReason());
// TODO: 触发退款流程（如果已预付）
// refundService.refund(order);
}
orderBasicInfoService.updateByPrimaryKeySelective(updateOrder);
return JcppUplinkResponse.success(message.getMessageId(), null);
} catch (Exception e) {
log.error("处理远程启动结果异常, tradeNo: {}, error: {}", tradeNo, e.getMessage(), e);
return JcppUplinkResponse.error(message.getMessageId(), "处理远程启动结果异常: " + e.getMessage());
}
}
@Override
public JcppUplinkResponse handleRemoteStopResult(JcppUplinkMessage message) {
String pileCode = message.getPileCode();
log.info("处理远程停止结果消息, pileCode: {}, sessionId: {}", pileCode, message.getSessionId());
// 远程停止结果通常不需要特殊处理，等待交易记录上报即可
// 这里只记录日志
return JcppUplinkResponse.success(message.getMessageId(), null);
}
/**
* 解析消息数据
*/
private <T> T parseData(Object data, Class<T> clazz) {
if (data == null) {
return null;
}
try {
if (data instanceof String) {
return JSON.parseObject((String) data, clazz);
}
return JSON.parseObject(JSON.toJSONString(data), clazz);
} catch (Exception e) {
log.error("解析消息数据失败, data: , clazz: {}, error: {}", data, clazz.getName(), e.getMessage());
return null;
}
}
}

179
jsowell-pile/src/main/java/com/jsowell/pile/jcpp/service/impl/JcppRemoteChargeServiceImpl.java Normal file
View File

@@ -0,0 +1,179 @@
package com.jsowell.pile.jcpp.service.impl;
import com.alibaba.fastjson2.JSON;
import com.jsowell.common.util.id.IdUtils;
import com.jsowell.pile.domain.MemberBasicInfo;
import com.jsowell.pile.domain.MemberWalletInfo;
import com.jsowell.pile.domain.OrderBasicInfo;
import com.jsowell.pile.domain.PileBasicInfo;
import com.jsowell.pile.jcpp.constant.JcppConstants;
import com.jsowell.pile.jcpp.dto.JcppSessionInfo;
import com.jsowell.pile.jcpp.service.IJcppDownlinkService;
import com.jsowell.pile.jcpp.service.IJcppRemoteChargeService;
import com.jsowell.pile.service.MemberBasicInfoService;
import com.jsowell.pile.service.MemberWalletInfoService;
import com.jsowell.pile.service.OrderBasicInfoService;
import com.jsowell.pile.service.PileBasicInfoService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.math.BigDecimal;
import java.util.Date;
import java.util.concurrent.TimeUnit;
/**
* JCPP 远程充电服务实现类
*
* @author jsowell
*/
@Slf4j
@Service
public class JcppRemoteChargeServiceImpl implements IJcppRemoteChargeService {
@Autowired
private PileBasicInfoService pileBasicInfoService;
@Autowired
private MemberBasicInfoService memberBasicInfoService;
@Autowired
private MemberWalletInfoService memberWalletInfoService;
@Autowired
private OrderBasicInfoService orderBasicInfoService;
@Autowired
private IJcppDownlinkService jcppDownlinkService;
@Autowired
private StringRedisTemplate stringRedisTemplate;
@Override
@Transactional(rollbackFor = Exception.class)
public String remoteStartCharging(String memberId, String pileCode, String gunNo, String payAmount) {
log.info("远程启动充电, memberId: {}, pileCode: {}, gunNo: {}, payAmount: {}",
memberId, pileCode, gunNo, payAmount);
// 1. 检查充电桩是否在线（使用新的 Redis Set 方式）
if (!jcppDownlinkService.isPileOnline(pileCode)) {
throw new RuntimeException("充电桩离线，请稍后重试");
}
// 2. 查询充电桩信息
PileBasicInfo pileInfo = pileBasicInfoService.selectPileBasicInfoBySN(pileCode);
if (pileInfo == null) {
throw new RuntimeException("充电桩不存在");
}
// 3. 查询会员信息
MemberBasicInfo memberInfo = memberBasicInfoService.selectInfoByMemberId(memberId);
if (memberInfo == null) {
throw new RuntimeException("会员不存在");
}
// 4. 检查会员状态
if ("1".equals(memberInfo.getStatus())) {
throw new RuntimeException("会员账户已冻结");
}
// 5. 查询钱包余额
String merchantId = String.valueOf(pileInfo.getMerchantId());
MemberWalletInfo walletInfo = memberWalletInfoService.selectByMemberId(memberId, merchantId);
BigDecimal balance = BigDecimal.ZERO;
if (walletInfo != null) {
BigDecimal principalBalance = walletInfo.getPrincipalBalance() != null ? walletInfo.getPrincipalBalance() : BigDecimal.ZERO;
BigDecimal giftBalance = walletInfo.getGiftBalance() != null ? walletInfo.getGiftBalance() : BigDecimal.ZERO;
balance = principalBalance.add(giftBalance);
}
// 6. 检查余额是否充足
BigDecimal payAmountDecimal = new BigDecimal(payAmount);
if (balance.compareTo(payAmountDecimal) < 0) {
throw new RuntimeException("余额不足");
}
// 7. 生成订单
String tradeNo = IdUtils.fastSimpleUUID();
OrderBasicInfo order = new OrderBasicInfo();
order.setOrderCode(tradeNo);
order.setTransactionCode(tradeNo);
order.setPileSn(pileCode);
order.setConnectorCode(pileCode + "-" + gunNo);
order.setMemberId(memberId);
order.setMerchantId(String.valueOf(pileInfo.getMerchantId()));
order.setStationId(String.valueOf(pileInfo.getStationId()));
order.setOrderStatus("0"); // 待充电（等待启动结果）
order.setPayStatus("0"); // 待支付
order.setChargeStartTime(new Date());
order.setCreateTime(new Date());
order.setPayAmount(payAmountDecimal);
orderBasicInfoService.insert(order);
// 8. 保存交易流水号映射（用于后续查询）
String tradeKey = "jcpp:trade:" + tradeNo;
stringRedisTemplate.opsForValue().set(tradeKey, pileCode, 24, TimeUnit.HOURS);
// 9. 发送远程启动指令（使用新的 Redis 队列方式）
jcppDownlinkService.sendRemoteStartCharging(pileCode, gunNo, tradeNo, payAmount, null, null);
log.info("远程启动充电指令已发送, tradeNo: {}, pileCode: {}, gunNo: {}", tradeNo, pileCode, gunNo);
return tradeNo;
}
@Override
public boolean remoteStopCharging(String memberId, String orderCode) {
log.info("远程停止充电, memberId: {}, orderCode: {}", memberId, orderCode);
// 1. 查询订单
OrderBasicInfo order = orderBasicInfoService.getOrderInfoByOrderCode(orderCode);
if (order == null) {
throw new RuntimeException("订单不存在");
}
// 2. 验证会员
if (!memberId.equals(order.getMemberId())) {
throw new RuntimeException("无权操作此订单");
}
// 3. 检查订单状态
if (!"1".equals(order.getOrderStatus())) {
throw new RuntimeException("订单状态不允许停止充电");
}
String pileCode = order.getPileSn();
String gunNo = order.getConnectorCode().replace(pileCode + "-", "");
// 4. 检查充电桩是否在线
if (!jcppDownlinkService.isPileOnline(pileCode)) {
throw new RuntimeException("充电桩离线，请稍后重试");
}
// 5. 发送远程停止指令（使用新的 Redis 队列方式）
jcppDownlinkService.sendRemoteStopCharging(pileCode, gunNo);
log.info("远程停止充电指令已发送, orderCode: {}, pileCode: {}, gunNo: {}", orderCode, pileCode, gunNo);
return true;
}
@Override
public boolean isPileOnline(String pileCode) {
// 使用新的 Redis Set 方式检查在线状态
return jcppDownlinkService.isPileOnline(pileCode);
}
@Override
public String getSessionId(String pileCode) {
String sessionKey = JcppConstants.REDIS_KEY_SESSION + pileCode;
String sessionJson = stringRedisTemplate.opsForValue().get(sessionKey);
if (sessionJson != null) {
JcppSessionInfo sessionInfo = JSON.parseObject(sessionJson, JcppSessionInfo.class);
return sessionInfo.getSessionId();
}
return null;
}
}