1

zj

2024-10-29 c8b1fc4dd96f24215cb5ed633a5b5c59c405f910

 src/main/java/com/nq/ws/USWebsocketRunClient.java

@@ -1,47 +1,130 @@
package com.nq.ws;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.gson.Gson;
import com.nq.constant.StockConstant;
import com.google.gson.reflect.TypeToken;
import com.nq.enums.EStockType;
import com.nq.pojo.StockRealTimeBean;
import com.nq.utils.redis.RedisKeyUtil;
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 java.io.IOException;
import java.lang.reflect.Type;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * @program: dabaogp
 * @description:
 * @create: 2024-10-25 14:44
 **/
@Slf4j
public class USWebsocketRunClient extends WebSocketClient {
public class USWebsocketRunClient  extends WebSocketClient {

    public USWebsocketRunClient(URI serverUri) {
    private EStockType eStockType;
    public USWebsocketRunClient(URI serverUri,
                              EStockType eStockType
    ) {
        super(serverUri);
        this.eStockType = eStockType;
    }

    @Override
    public void onOpen(ServerHandshake serverHandshake) {
        log.info("webSocket 客服端链接成功");
        send(("key:"+ EStockType.US.stockKey+":"+EStockType.US.getContryId()).getBytes());
        send(("key:"+ eStockType.getStockKey()+":"+eStockType.getContryId()).getBytes());
    }

    @Override
    public void onMessage(String s) {
        try {
            log.info("美股实时数据 {}",s);
            StockRealTimeBean stockDetailBean =  new Gson().fromJson(s, StockRealTimeBean.class);
            RedisKeyUtil.setCacheRealTimeStock(EStockType.US,stockDetailBean);
            if(!s.equals("pong") && !s.equals("身份验证成功")){
                Map<String, String> stringObjectMap = jsonToMap(s);
                StockRealTimeBean stockRealTimeBean = new StockRealTimeBean();
                stockRealTimeBean.setPid(stringObjectMap.get("Id").toString());
                stockRealTimeBean.setLast(stringObjectMap.get("Last").toString());
                stockRealTimeBean.setBid(stringObjectMap.get("Bid").toString());
                stockRealTimeBean.setAsk(stringObjectMap.get("Ask").toString());
                stockRealTimeBean.setHigh(stringObjectMap.get("High").toString());
                stockRealTimeBean.setLow(stringObjectMap.get("Low").toString());
                stockRealTimeBean.setPc(stringObjectMap.get("Chg").toString());
                stockRealTimeBean.setPcp(stringObjectMap.get("ChgPct").toString()+"%");
                stockRealTimeBean.setTime(stringObjectMap.get("Time").toString());
                ObjectMapper objectMapper = new ObjectMapper();
                try {
                    if(!stockRealTimeBean.getPcp().contains("-")){
                        stockRealTimeBean.setPcp("+"+stringObjectMap.get("ChgPct").toString()+"%");
                    }
                    String json = objectMapper.writeValueAsString(stockRealTimeBean);
                    RedisKeyUtil.setCacheRealTimeStock(eStockType,stockRealTimeBean);
                    send(json);
                } catch (
                        JsonProcessingException e) {
                    throw new RuntimeException(e);
                }
            }
        }catch (Exception e){

            log.error("ws 消息接收错误："+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 关闭"+1);
        log.info("websocket  美股  关闭"+1);
    }

    @Override
    public void onError(Exception e) {
        log.info("websocket 错误");
    }


    private HttpClient httpClient;  //  声明一个成员变量

    private HttpPost httpPost = new HttpPost("http://127.0.0.1:8002/api/sendNotification");  //  创建一个HttpPost对象

    @Override
    public void send(String message) {

        try {

            if (httpClient == null) {
                httpClient = HttpClients.createDefault();  //  使用单例模式创建HttpClient对象
            }

            //  准备  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) {
            e.printStackTrace();
        }
    }

}