From dc9bd22833255bc602dd42c7f603ecb50842ab35 Mon Sep 17 00:00:00 2001
From: zj <1772600164@qq.com>
Date: Mon, 23 Jun 2025 19:11:16 +0800
Subject: [PATCH] 1
---
src/main/java/project/contract/job/ContractOrderCalculationServiceImpl.java | 141 +++++++++++++++++++++++++++++++----------------
1 files changed, 93 insertions(+), 48 deletions(-)
diff --git a/src/main/java/project/contract/job/ContractOrderCalculationServiceImpl.java b/src/main/java/project/contract/job/ContractOrderCalculationServiceImpl.java
index 5a8b9f9..4b26b2e 100644
--- a/src/main/java/project/contract/job/ContractOrderCalculationServiceImpl.java
+++ b/src/main/java/project/contract/job/ContractOrderCalculationServiceImpl.java
@@ -1,10 +1,15 @@
package project.contract.job;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
import java.text.DecimalFormat;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
import kernel.util.StringUtils;
+import kernel.web.ApplicationUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -14,6 +19,11 @@
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
+import org.springframework.jdbc.core.BatchPreparedStatementSetter;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
+import org.springframework.transaction.TransactionStatus;
+import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.util.ObjectUtils;
import project.contract.ContractLock;
import project.contract.ContractOrder;
@@ -27,6 +37,7 @@
import project.wallet.Wallet;
import project.wallet.WalletRedisKeys;
import project.wallet.WalletService;
+import project.wallet.consumer.WalletDao;
import project.wallet.consumer.WalletMessage;
public class ContractOrderCalculationServiceImpl implements ContractOrderCalculationService, ApplicationContextAware {
@@ -36,6 +47,7 @@
private ContractOrderService contractOrderService;
private DataService dataService;
private WalletService walletService;
+ private WalletDao walletDao;
// private RedisHandler redisHandler;
private AssetService assetService;
public final static String STATE_SUBMITTED = "submitted";
@@ -55,6 +67,7 @@
try {
ContractOrder order = contractOrderService.findByOrderNo(order_no);
if (order == null || !ContractOrder.STATE_SUBMITTED.equals(order.getState())) {
+ logger.info("-----------到这里来了吗");
/**
* 状态已改变，退出处理
*/
@@ -62,6 +75,7 @@
}
List<Realtime> list = this.dataService.realtime(order.getSymbol());
if (list.size() == 0) {
+ logger.info("-----------去你妈退出了");
return;
}
Realtime realtime = list.get(0);
@@ -98,6 +112,9 @@
}
+ private static final Lock lock = new ReentrantLock(); // 全局锁，避免重复执行
+ private static final long SLEEP_TIME = 500; // 重试间隔
+
/**
* 盈亏计算
*
@@ -117,18 +134,28 @@
* 盈 正数
*/
order.setProfit(Arith.add(0.0D, Math.abs(amount)));
- redisHandler.setSync("MONEY_CONTRACT_PROFIT_"+order.getPartyId().toString(), 0);
+
+ Object profit = redisHandler.get("MONEY_CONTRACT_PROFIT_" + order.getPartyId().toString());
+ if(!ObjectUtils.isEmpty(profit)){
+ if(Double.parseDouble(profit.toString()) < 0 ){
+
+ redisHandler.setSync("MONEY_CONTRACT_PROFIT_"+order.getPartyId().toString(), 0);
+
+ Object money = redisHandler.get("PARTY_ID_MONEY_" + order.getPartyId().toString());
+ if(!ObjectUtils.isEmpty(money)){
+ Wallet wallet = this.walletService.saveWalletByPartyId(order.getPartyId().toString());
+ wallet.setMoney(Double.parseDouble(money.toString()));
+ redisHandler.setSync(WalletRedisKeys.WALLET_PARTY_ID + wallet.getPartyId().toString(), wallet);
+ redisHandler.pushAsyn(WalletRedisKeys.WALLET_QUEUE_UPDATE, new WalletMessage(order.getPartyId().toString(), amount));
+ }
+ }
+ }
} else if ("loss".equals(profit_loss)) {
order.setProfit(Arith.sub(0.0D,Math.abs(amount) ));
- //--------------------------------------------- 这一段代码是新加的，服务器上面现在是注释的
//定义一个 总浮动亏损的值
- logger.info("----------fffffffffffffff-------"+redisHandler);
- logger.info("----------ssssssssssss-------"+order.getPartyId().toString());
Double contractAssetsProfit = (Double) redisHandler.get(ContractRedisKeys.CONTRACT_ASSETS_PROFIT_PARTY_ID + order.getPartyId().toString());
- logger.info("----------bbbbbbbbbbbbbbbbbb-------"+contractAssetsProfit);
double contractProfit = null == contractAssetsProfit ? 0.000D : contractAssetsProfit;
- logger.info("----------contractProfit-------"+contractProfit);
List<ContractOrder> list = contractOrderService.findSubmitted(order.getPartyId().toString(), null, null);
double deposit = 0;
for (int i = 0; i < list.size(); i++) {
@@ -136,29 +163,23 @@
deposit = Arith.add(deposit, close_line.getDeposit());
}
double profitt = Arith.add(contractProfit,deposit);
- logger.info("----------盈亏除去保证金之后的值-------"+profitt);
if(profitt <= 0){
redisHandler.setSync("MONEY_CONTRACT_PROFIT_"+order.getPartyId().toString(), profitt);
}else{
redisHandler.setSync("MONEY_CONTRACT_PROFIT_"+order.getPartyId().toString(), 0);
}
- logger.info("----------1111111111111-------");
//修改余额,每次修改取平仓前余额减去浮动亏损
Wallet wallet = this.walletService.saveWalletByPartyId(order.getPartyId().toString());
- logger.info("----------wallet-------"+wallet);
Object money = redisHandler.get("PARTY_ID_MONEY_" + order.getPartyId().toString());
Object profit = redisHandler.get("MONEY_CONTRACT_PROFIT_" + order.getPartyId().toString());
- logger.info("----------money-------"+money);
- logger.info("----------profit-------"+profit);
if(!ObjectUtils.isEmpty(money) && !ObjectUtils.isEmpty(profit)){
- wallet.setMoney(Arith.add(Double.parseDouble(money.toString()),Double.parseDouble(profit.toString())));
+ double residueMoney = Arith.add(Double.parseDouble(money.toString()), Double.parseDouble(profit.toString()));
+ //如果当前余额减去浮动亏损小于0,将进入强平，这里不再进行余额更新
+ if(residueMoney > 0){
+ wallet.setMoney(residueMoney);
+ redisHandler.setSync(WalletRedisKeys.WALLET_PARTY_ID + wallet.getPartyId().toString(), wallet);
+ }
}
- logger.info("----------2222222222222-------");
- redisHandler.setSync(WalletRedisKeys.WALLET_PARTY_ID + wallet.getPartyId().toString(), wallet);
- logger.info("----------3333333333333333------");
- redisHandler.pushAsyn(WalletRedisKeys.WALLET_QUEUE_UPDATE, new WalletMessage(order.getPartyId().toString(), amount));
- logger.info("----------4444444444444-------");
- //------------------------------------------
}
double changeRatio;
@@ -231,47 +252,33 @@
profit = Arith.add(profit, Arith.add(close_line.getProfit(), close_line.getDeposit()));
}
Wallet wallet = this.walletService.saveWalletByPartyId(order.getPartyId().toString());
- double totleMoney = wallet.getMoney();
+ Object money = redisHandler.get("PARTY_ID_MONEY_" + order.getPartyId().toString());
+ double totleMoney = 0;
+ if(ObjectUtils.isEmpty(money)){
+ totleMoney = wallet.getMoney();
+ }else{
+ totleMoney = Double.parseDouble(money.toString());
+ }
+ logger.info("---------全仓强平任务----------------");
+ logger.info("-------------------------:"+Arith.add(profit,totleMoney));
if (Arith.add(profit,totleMoney) <= 0) {
+ redisHandler.setSync("MONEY_CONTRACT_PROFIT_"+order.getPartyId().toString(), 0);
/**
* 触发全仓强平
*/
this.contractOrderService.saveClose(order.getPartyId().toString(), order.getOrder_no());
- ThreadUtils.sleep(100);
for (int i = 0; i < list.size(); i++) {
ContractOrder close_line = list.get(i);
- if (!order.getOrder_no().equals(close_line.getOrder_no())) {
- try {
-
- while (true) {
- if (ContractLock.add(close_line.getOrder_no())) {
- this.contractOrderService.saveClose(close_line.getPartyId().toString(),
- close_line.getOrder_no());
- /**
- * 处理完退出
- */
- break;
- }
- ThreadUtils.sleep(500);
-
- }
-
- } catch (Exception e) {
- logger.error("error:", e);
- } finally {
- ContractLock.remove(close_line.getOrder_no());
- ThreadUtils.sleep(100);
- }
-
+ synchronized (close_line.getOrder_no()){
+ this.contractOrderService.saveClose(close_line.getPartyId().toString(),
+ close_line.getOrder_no());
}
}
-
+ //钱包归零
+ extracted(order, wallet, redisHandler);
}
} else {
logger.info("---------进入单个持仓----------------");
- logger.info("---------order.getProfit()----------------"+order.getProfit());
- logger.info("---------order.getDeposit()----------------"+order.getDeposit());
- logger.info("---------order_close_line----------------"+order_close_line);
if (order.getProfit() < 0 && (Arith.div(order.getDeposit(), Math.abs(order.getProfit())) <= Arith
.div(order_close_line, 100))) {
logger.info("---------进入强平了----------------");
@@ -282,8 +289,46 @@
return;
}
}
-
}
+
+ private void extracted(ContractOrder order, Wallet wallet, RedisHandler redisHandler) {
+ DataSourceTransactionManager transactionManager = ApplicationUtil.getBean(DataSourceTransactionManager.class);
+ JdbcTemplate jdbcTemplate = ApplicationUtil.getBean(JdbcTemplate.class);
+ TransactionStatus status = null;
+ try {
+ // 开启事务
+ status = transactionManager.getTransaction(new DefaultTransactionDefinition());
+
+ logger.info("----------我来更新钱包了");
+ wallet.setMoney(0);
+ redisHandler.setSync(WalletRedisKeys.WALLET_PARTY_ID + wallet.getPartyId().toString(), wallet);
+
+ logger.info("-------wallet:" + wallet.getPartyId());
+ // 更新数据库
+ int update = jdbcTemplate.update("UPDATE T_WALLET SET MONEY=ROUND(?,8) WHERE PARTY_ID=?", 0, wallet.getPartyId().toString());
+ logger.info("--------cnm跟新了几条：" + update);
+
+ // 保存钱包信息
+ Wallet ww = this.walletService.saveWalletByPartyId(order.getPartyId().toString());
+ logger.info("-----------到底他妈是什么值：" + ww.getMoney());
+
+ // 更新 redis
+ redisHandler.setSync("PARTY_ID_MONEY_" + wallet.getPartyId().toString(), wallet.getMoney());
+
+ // 提交事务
+ transactionManager.commit(status);
+ logger.info("----------更新完了");
+
+ } catch (Exception e) {
+ // 回滚事务
+ if (status != null) {
+ transactionManager.rollback(status);
+ }
+ logger.error("更新钱包时发生错误：", e);
+ throw e;
+ }
+ }
+
public void setDataService(DataService dataService) {
this.dataService = dataService;
}
--
Gitblit v1.9.3