WebSocketClient: A Java Class for Real-Time Data Acquisition and WebSocket Communication
@Slf4j
@Component
public class WebSocketClient {
private static WebSocket webSocket = null;
private static Boolean flag = false;
private static Boolean isConnect = false;
private static String sign;
private final static HashFunction crc32 = Hashing.crc32();
private final static ObjectReader objectReader = new ObjectMapper().readerFor(OrderBookData.class);
private static Map<String, Optional<SpotOrderBook>> bookMap = new HashMap<>();
public WebSocketClient() {
}
@Resource
private RealTimeDataAcquisitionService realTimeDataAcquisitionService;
private static WebSocketClient socketClient;
@PostConstruct
public void init() {
socketClient = this;
socketClient.realTimeDataAcquisitionService = this.realTimeDataAcquisitionService;
}
public static void setRedis(String x) {
socketClient.realTimeDataAcquisitionService.realTimeDataAcquisition1s(x);
}
//与服务器建立连接，参数为服务器的URL
public static WebSocket connection(final String url) {
OkHttpClient client = new OkHttpClient.Builder()
.readTimeout(5, TimeUnit.SECONDS)
.build();
Request request = new Request.Builder()
.url(url)
.build();
webSocket = client.newWebSocket(request, new WebSocketListener() {
ScheduledExecutorService service;
@Override
public void onOpen(final WebSocket webSocket, final Response response) {
//连接成功后，设置定时器，每隔25s，自动向服务器发送心跳，保持与服务器连接
isConnect = true;
System.out.println(Instant.now().toString() + ' Connected to the server success!');
Runnable runnable = new Runnable() {
public void run() {
// task to run goes here
sendMessage('ping');
}
};
service = Executors.newSingleThreadScheduledExecutor();
// 第二个参数为首次执行的延时时间，第三个参数为定时执行的间隔时间
service.scheduleAtFixedRate(runnable, 25, 25, TimeUnit.SECONDS);
}
@Override
public void onClosing(WebSocket webSocket, int code, String reason) {
System.out.println('Connection is about to disconnect!');
webSocket.close(1000, 'Long time no message was sent or received!');
webSocket = null;
}
@Override
public void onClosed(final WebSocket webSocket, final int code, final String reason) {
System.out.println('Connection dropped!');
}
@Override
public void onFailure(final WebSocket webSocket, final Throwable t, final Response response) {
System.out.println('Connection failed,Please reconnect!');
System.out.println('订阅失败的参数：' + t.getMessage());
System.out.println('参数：' + webSocket.request());
if (Objects.nonNull(service)) {
service.shutdown();
reConnect();
}
}
@Override
public void onMessage(final WebSocket webSocket, final String s) {
setRedis(s);
if(s.contains('pong')){
System.out.println(DateFormatUtils.format(new Date(), DateUtils.TIME_STYLE_S4) + ' Receive: ' + s);
}
if (null != s && s.contains('login')) {
if (s.endsWith('true}')) {
flag = true;
}
}
}
});
return webSocket;
}
public static void reConnect() {
log.info('重连:{};{}');
try {
WebSocketConfig.publicConnect(socketClient);
//开启订阅
socketClient.realTimeDataAcquisitionService.reSubscribe();
} catch (Exception e) {
e.printStackTrace();
}
}
private static void isLogin(String s) {
if (null != s && s.contains('login')) {
if (s.endsWith('true}')) {
flag = true;
}
}
}
//获得sign
private static String sha256_HMAC(String message, String secret) {
String hash = '';
try {
Mac sha256_HMAC = Mac.getInstance('HmacSHA256');
SecretKeySpec secret_key = new SecretKeySpec(secret.getBytes(CharsetEnum.UTF_8.charset()), 'HmacSHA256');
sha256_HMAC.init(secret_key);
byte[] bytes = sha256_HMAC.doFinal(message.getBytes(CharsetEnum.UTF_8.charset()));
hash = Base64.getEncoder().encodeToString(bytes);
} catch (Exception e) {
System.out.println('Error HmacSHA256 ===========' + e.getMessage());
}
return hash;
}
private static String listToJson(List<Map> list) {
JSONArray jsonArray = new JSONArray();
for (Map map : list) {
jsonArray.add(JSONObject.fromObject(map));
}
return jsonArray.toJSONString();
}
//登录
public static void login(String apiKey, String passPhrase, String secretKey) {
String timestamp = (Double.parseDouble(DateUtils.getEpochTime()) + 28800) + '';
String message = timestamp + 'GET' + '/users/self/verify';
sign = sha256_HMAC(message, secretKey);
String str = '{\'op\'' + ':' + '\'login\'' + ',' + '\'args\'' + ':' + '[' + '\'' + apiKey + '\'' + ',' + '\'' + passPhrase + '\'' + ',' + '\'' + timestamp + '\'' + ',' + '\'' + sign + '\'' + ']}';
sendMessage(str);
}
//订阅，参数为频道组成的集合
public static void subscribe(List<Map> list) {
// String s = listToJson(list);
// String str = '{\'op\': \'subscribe\', \'args\':' + s + '}';
Map<String,Object> map=new HashMap<>();
map.put('op','subscribe');
map.put('args',list);
String str = JSONArray.toJSONString(map);
if (null != webSocket)
sendMessage(str);
}
//取消订阅，参数为频道组成的集合
public static void unsubscribe(List<Map> list) {
String s = listToJson(list);
String str = '{\'op\': \'unsubscribe\', \'args\':' + s + '}';
if (null != webSocket)
sendMessage(str);
}
private static void sendMessage(String str) {
if (null != webSocket) {
try {
Thread.sleep(1300);
} catch (Exception e) {
e.printStackTrace();
}
System.out.println(DateFormatUtils.format(new Date(), DateUtils.TIME_STYLE_S4) + 'Send a message to the server:' + str);
webSocket.send(str);
} else {
System.out.println('Please establish the connection before you operate it!');
}
}
//断开连接
public static void closeConnection() {
if (null != webSocket) {
webSocket.close(1000, 'User actively closes the connection');
} else {
System.out.println('Please establish the connection before you operate it!');
}
}
public boolean getIsLogin() {
return flag;
}
public boolean getIsConnect() {
return isConnect;
}
public static <T extends OrderBookItem> int checksum(List<T> asks, List<T> bids) {
System.out.println('深度');
StringBuilder s = new StringBuilder();
for (int i = 0; i < 25; i++) {
if (i < bids.size()) {
s.append(bids.get(i).getPrice().toString());
s.append(':');
s.append(bids.get(i).getSize());
s.append(':');
}
if (i < asks.size()) {
s.append(asks.get(i).getPrice().toString());
s.append(':');
s.append(asks.get(i).getSize());
s.append(':');
}
}
final String str;
if (s.length() > 0) {
str = s.substring(0, s.length() - 1);
} else {
str = '';
}
return crc32.hashString(str, StandardCharsets.UTF_8).asInt();
}
private static <T extends OrderBookItem> String getStr(List<T> asks, List<T> bids) {
StringBuilder s = new StringBuilder();
for (int i = 0; i < 25; i++) {
if (i < bids.size()) {
s.append(bids.get(i).getPrice().toString());
s.append(':');
s.append(bids.get(i).getSize());
s.append(':');
}
if (i < asks.size()) {
s.append(asks.get(i).getPrice().toString());
s.append(':');
s.append(asks.get(i).getSize());
s.append(':');
}
}
final String str;
if (s.length() > 0) {
str = s.substring(0, s.length() - 1);
} else {
str = '';
}
return str;
}
public static Optional<SpotOrderBook> parse(String json) {
try {
OrderBookData data = objectReader.readValue(json);
List<SpotOrderBookItem> asks =
data.getAsks().stream().map(x -> new SpotOrderBookItem(new String(x.get(0)), x.get(1), x.get(2), x.get(3)))
.collect(Collectors.toList());
List<SpotOrderBookItem> bids =
data.getBids().stream().map(x -> new SpotOrderBookItem(new String(x.get(0)), x.get(1), x.get(2), x.get(3)))
.collect(Collectors.toList());
return Optional.of(new SpotOrderBook(asks, bids, data.getTs(), data.getChecksum()));
} catch (Exception e) {
return Optional.empty();
}
}
@Data
public static class OrderBookData {
private List<List<String>> asks;
private List<List<String>> bids;
private String ts;
private int checksum;
public List<List<String>> getAsks() {
return asks;
}
public void setAsks(List<List<String>> asks) {
this.asks = asks;
}
public List<List<String>> getBids() {
return bids;
}
public void setBids(List<List<String>> bids) {
this.bids = bids;
}
public String getTs() {
return ts;
}
public void setTs(String ts) {
this.ts = ts;
}
public int getChecksum() {
return checksum;
}
public void setChecksum(int checksum) {
this.checksum = checksum;
}
}
}
WebSocketClient指定线程池配置websocketAsyncExecutor
This Java code snippet demonstrates a WebSocketClient class, designed to manage real-time data acquisition through a WebSocket connection. The class is responsible for:
-
Establishing a WebSocket connection: The
connection()method initializes an OkHttpClient instance with a configured read timeout and builds a request object using the provided URL. The connection is established by creating a new WebSocket instance and providing a WebSocketListener to handle events like opening, closing, failures, and incoming messages. -
Managing the connection: The WebSocketListener includes methods like
onOpen(),onClosing(),onClosed(),onFailure(), andonMessage(), which respond to different WebSocket events. For example,onOpen()sets a timer to periodically send heartbeats to maintain the connection, whileonFailure()initiates a reconnection attempt if needed. -
Sending messages: The
sendMessage()method transmits messages over the established WebSocket connection. -
Handling incoming messages: The
onMessage()method processes incoming messages, extracting relevant data and sending it to a Redis instance. -
Login and subscription: The
login()method handles user authentication with the server, and thesubscribe()method subscribes to specific channels to receive real-time data feeds. -
Data parsing: The
parse()method converts incoming JSON data into aSpotOrderBookobject, which represents the order book for a particular asset.
The WebSocketClient class provides various utility methods to manage the connection lifecycle, handle messages, and interact with the real-time data. It is designed to integrate with a larger system responsible for data acquisition, processing, and storage.
Please note that this code snippet is incomplete and assumes the existence of supporting classes like RealTimeDataAcquisitionService, WebSocketConfig, OrderBookItem, SpotOrderBookItem, SpotOrderBook, OrderBookData, and others. You will need to define these classes to use this code snippet effectively.
原文地址: https://www.cveoy.top/t/topic/o8jj 著作权归作者所有。请勿转载和采集!