package com.yami.trading.huobi.jsws;
|
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
import com.google.gson.Gson;
|
import com.google.gson.reflect.TypeToken;
|
|
import com.yami.trading.bean.data.domain.Realtime;
|
import com.yami.trading.bean.item.domain.Item;
|
import com.yami.trading.bean.model.StockWSBean;
|
import com.yami.trading.huobi.data.DataCache;
|
import com.yami.trading.huobi.websocket.constant.enums.EStockType;
|
import com.yami.trading.service.item.ItemService;
|
import lombok.extern.slf4j.Slf4j;
|
import org.apache.http.HttpResponse;
|
import org.apache.http.client.HttpClient;
|
import org.apache.http.client.entity.UrlEncodedFormEntity;
|
import org.apache.http.client.methods.HttpPost;
|
import org.apache.http.impl.client.HttpClients;
|
import org.apache.http.message.BasicNameValuePair;
|
import org.java_websocket.client.WebSocketClient;
|
import org.java_websocket.handshake.ServerHandshake;
|
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.context.annotation.Scope;
|
import org.springframework.stereotype.Component;
|
|
import java.io.IOException;
|
import java.lang.reflect.Type;
|
import java.math.BigDecimal;
|
import java.math.RoundingMode;
|
import java.net.URI;
|
import java.nio.charset.StandardCharsets;
|
import java.util.*;
|
|
@Slf4j
|
@Component
|
@Scope("prototype")
|
public class WebsocketRunClient extends WebSocketClient {
|
|
private static HttpClient httpClient = HttpClients.createDefault(); // 单例化 HttpClient
|
|
|
private static HttpPost httpPost;
|
static {
|
httpPost = new HttpPost("http://127.0.0.1:8001/api/sendNotification"); // 初始化 HttpPost
|
}
|
|
private EStockType eStockType;
|
|
@Autowired
|
ItemService itemService;
|
|
public WebsocketRunClient() {
|
super(URI.create(EStockType.US.wsUrl));
|
}
|
|
public WebsocketRunClient(URI serverUri, EStockType eStockType) {
|
// 修改为新的WebSocket服务器地址
|
super(URI.create(eStockType.wsUrl));
|
this.eStockType = eStockType;
|
}
|
|
|
@Override
|
public void onOpen(ServerHandshake serverHandshake) {
|
log.info("WebSocket连接已建立，连接到: {}", EStockType.US.wsUrl);
|
// 发送身份验证消息
|
send(("key:"+ eStockType.getStockKey()+":"+eStockType.getContryId()).getBytes());
|
Timer heartbeatTimer;
|
// 启动心跳定时器
|
heartbeatTimer = new Timer();
|
heartbeatTimer.schedule(new TimerTask() {
|
@Override
|
public void run() {
|
if (isOpen()) {
|
//send(("key:"+ eStockType.getStockKey()+":"+eStockType.getContryId()).getBytes());
|
send("heartbeat");
|
}
|
}
|
}, 0, 3000); // 每3秒发送一次心跳消息
|
}
|
|
@Override
|
public void onMessage(String s) {
|
if(!s.equals("pong") && !s.equals("身份验证成功")){
|
try {
|
Map<String, String> stringObjectMap = jsonToMap(s);
|
String symbol = stringObjectMap.get("Symbol").trim();
|
Item item = itemService.findCaCheBySymbol(symbol);
|
if (item == null || !item.getType().equalsIgnoreCase(Item.US_STOCKS)) {
|
return;
|
}
|
System.out.println(s);
|
StockWSBean stockRealTimeBean = new StockWSBean();
|
stockRealTimeBean.setPid(stringObjectMap.get("Id"));
|
stockRealTimeBean.setLast(stringObjectMap.get("Last"));
|
stockRealTimeBean.setBid(stringObjectMap.get("Bid"));
|
stockRealTimeBean.setAsk(stringObjectMap.get("Ask"));
|
stockRealTimeBean.setHigh(stringObjectMap.get("High"));
|
stockRealTimeBean.setLow(stringObjectMap.get("Low"));
|
stockRealTimeBean.setPc(stringObjectMap.get("Chg"));
|
stockRealTimeBean.setPcp(stringObjectMap.get("ChgPct") +"%");
|
stockRealTimeBean.setTime(stringObjectMap.get("Time"));
|
stockRealTimeBean.setOpen(stringObjectMap.get("Open"));
|
stockRealTimeBean.setPrevClose(stringObjectMap.get("PrevClose"));
|
stockRealTimeBean.setSymbol(stringObjectMap.get("Symbol"));
|
Realtime realtime = new Realtime();
|
/*realtime.setUuid(stringObjectMap.get("pid"));
|
realtime.setSymbol(stringObjectMap.get("symbol"));
|
realtime.setClose(new BigDecimal(stringObjectMap.get("last")).doubleValue());
|
realtime.setLow(new BigDecimal(stringObjectMap.get("low")).doubleValue());
|
realtime.setHigh(new BigDecimal(stringObjectMap.get("high")).doubleValue());
|
realtime.setOpen(new BigDecimal(stringObjectMap.get("open")).doubleValue());
|
realtime.setPrevClose(new BigDecimal(stringObjectMap.get("prevClose")).doubleValue());
|
realtime.setTs(Long.valueOf(stringObjectMap.get("time") + "000"));
|
realtime.setNetChange(new BigDecimal(stringObjectMap.get("pc")).doubleValue());
|
realtime.setChangeRatio(parsePercent(stringObjectMap.get("pcp")));
|
realtime.setBid(new BigDecimal(stringObjectMap.get("bid")).doubleValue());
|
realtime.setAsk(new BigDecimal(stringObjectMap.get("ask")).doubleValue());*/
|
realtime.setUuid(stringObjectMap.get("Id"));
|
realtime.setSymbol(stringObjectMap.get("Symbol"));
|
realtime.setClose(new BigDecimal(stringObjectMap.get("Last")).doubleValue());
|
realtime.setLow(new BigDecimal(stringObjectMap.get("Low")).doubleValue());
|
realtime.setHigh(new BigDecimal(stringObjectMap.get("High")).doubleValue());
|
realtime.setOpen(new BigDecimal(stringObjectMap.get("Open")).doubleValue());
|
realtime.setPrevClose(new BigDecimal(stringObjectMap.get("PrevClose")).doubleValue());
|
realtime.setTs(Long.valueOf(stringObjectMap.get("Time") + "000"));
|
realtime.setNetChange(new BigDecimal(stringObjectMap.get("Chg")).doubleValue());
|
realtime.setChangeRatio(new BigDecimal(stringObjectMap.get("ChgPct")).doubleValue());
|
realtime.setBid(new BigDecimal(stringObjectMap.get("Bid")).doubleValue());
|
realtime.setAsk(new BigDecimal(stringObjectMap.get("Ask")).doubleValue());
|
|
|
DataCache.putRealtime(realtime.getSymbol(), realtime);
|
|
ObjectMapper objectMapper = new ObjectMapper();
|
String json = objectMapper.writeValueAsString(stockRealTimeBean);
|
sendLoca(json);
|
|
} catch (Exception e) {
|
log.error("处理WebSocket消息时发生错误: {}", e);
|
}
|
} else {
|
log.info("WebSocket心跳或认证响应: {}", s);
|
}
|
}
|
|
public static Map<String, String> jsonToMap(String json) {
|
Gson gson = new Gson();
|
Type type = new TypeToken<Map<String, String>>(){}.getType();
|
return gson.fromJson(json, type);
|
}
|
|
@Override
|
public void onClose(int i, String s, boolean b) {
|
log.info("WebSocket连接已关闭，代码: {}, 原因: {}, 远程关闭: {}", i, s, b);
|
}
|
|
@Override
|
public void onError(Exception e) {
|
log.error("WebSocket连接发生错误: {}", e.getMessage(), e);
|
}
|
|
/**
|
* 将带正负号的百分比字符串转换为double（返回小数形式，如+37.69% → 0.3769）
|
* @param percentStr 百分比字符串（如"+37.69%"、"-5.2%"）
|
* @return 转换后的double值
|
* @throws IllegalArgumentException 若输入格式无效
|
*/
|
public static double parsePercent(String percentStr) {
|
if (percentStr == null || percentStr.trim().isEmpty()) {
|
throw new IllegalArgumentException("输入字符串不能为空");
|
}
|
|
// 1. 去除百分号并清理空格
|
String numStr = percentStr.replace("%", "").trim();
|
|
try {
|
// 2. 转换为double（支持正负号）
|
double value = Double.parseDouble(numStr);
|
// 3. 转换为小数（百分比 → 小数）
|
double decimal = value / 100.0;
|
// 4. 四舍五入保留4位小数（使用BigDecimal确保精度）
|
return new BigDecimal(decimal)
|
.setScale(4, RoundingMode.HALF_UP) // HALF_UP：四舍五入模式
|
.doubleValue();
|
} catch (NumberFormatException e) {
|
throw new IllegalArgumentException("无效的百分比格式：" + percentStr, e);
|
}
|
}
|
|
public boolean verify(String str) {
|
if (str == null) {
|
return false;
|
}
|
if (str.trim().isEmpty()) {
|
return false;
|
}
|
if (str.trim().equals("undefined")) {
|
return false;
|
}
|
if (str.trim().equals("null")) {
|
return false;
|
}
|
return true;
|
}
|
|
public void sendLoca(String message) {
|
try {
|
// 准备 form-data 参数
|
List<BasicNameValuePair> params = new ArrayList<>();
|
params.add(new BasicNameValuePair("message", message));
|
|
// 设置编码格式为 UTF-8
|
UrlEncodedFormEntity entity = new UrlEncodedFormEntity(params, StandardCharsets.UTF_8);
|
httpPost.setEntity(entity); // 设置 HttpPost 对象的参数
|
|
// 发送请求
|
HttpResponse response = httpClient.execute(httpPost);
|
|
// 处理响应
|
int statusCode = response.getStatusLine().getStatusCode();
|
} catch (IOException e) {
|
log.error("Http 请求错误", e);
|
}
|
}
|
}
|
