新增websocket通信（中继服务同步相关）
anthonywj
2 years ago
parent
2314e8076b
commit
ff58a80153
@ -0,0 +1,12 @@
|
||||
package com.glxp.sale.admin.config;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
|
||||
|
||||
@Configuration
|
||||
public class WebSocketConfig {
|
||||
@Bean
|
||||
public ServerEndpointExporter serverEndpointExporter() {
|
||||
return new ServerEndpointExporter();
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,7 @@
|
||||
package com.glxp.sale.admin.constant;
|
||||
|
||||
public interface SocketMsgType {
|
||||
|
||||
|
||||
String DL_ALL_DATA = "DL_ALL_DATA";
|
||||
}
|
||||
@ -0,0 +1,19 @@
|
||||
package com.glxp.sale.admin.entity.sync;
|
||||
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Builder;
|
||||
import lombok.Data;
|
||||
import lombok.NoArgsConstructor;
|
||||
|
||||
/**
|
||||
* websocket 消息体
|
||||
*/
|
||||
@Data
|
||||
@Builder
|
||||
@AllArgsConstructor
|
||||
@NoArgsConstructor
|
||||
public class SocketMsgEntity {
|
||||
private String type;
|
||||
private String content;
|
||||
private String remark;
|
||||
}
|
||||
@ -0,0 +1,20 @@
|
||||
package com.glxp.sale.admin.socket.client;
|
||||
|
||||
import com.glxp.sale.admin.socket.server.SpsSyncWebSocket;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import javax.annotation.Resource;
|
||||
|
||||
@Service
|
||||
public class SocketMsgService {
|
||||
|
||||
@Value("${SPMS_WEBSOCKET_TOKEN}")
|
||||
String token;
|
||||
@Resource
|
||||
SpsSyncWebSocket spsSyncWebSocket;
|
||||
|
||||
public void sendNoticeMsg(String message) {
|
||||
spsSyncWebSocket.sendMessage(message, "2:" + token);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,113 @@
|
||||
package com.glxp.sale.admin.socket.client;
|
||||
|
||||
import com.alibaba.fastjson.JSONObject;
|
||||
import com.glxp.sale.admin.constant.SocketMsgType;
|
||||
import org.java_websocket.WebSocket;
|
||||
import org.java_websocket.client.WebSocketClient;
|
||||
import org.java_websocket.framing.Framedata;
|
||||
import org.java_websocket.handshake.ServerHandshake;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import javax.annotation.Resource;
|
||||
import java.net.URI;
|
||||
import java.net.URISyntaxException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
|
||||
public class SpsWebSocketClient extends WebSocketClient {
|
||||
|
||||
private String excptMessage;
|
||||
|
||||
Logger log = LoggerFactory.getLogger(SpsWebSocketClient.class);
|
||||
|
||||
private static List<SpsWebSocketClient> list = new ArrayList<>();
|
||||
@Resource
|
||||
SocketMsgService socketMsgService;
|
||||
|
||||
public void setSocketMsgService(SocketMsgService socketMsgService) {
|
||||
this.socketMsgService = socketMsgService;
|
||||
}
|
||||
|
||||
public SpsWebSocketClient(String serverUri) throws URISyntaxException {
|
||||
super(new URI(serverUri));
|
||||
this.setConnectionLostTimeout(0);
|
||||
if (list.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
for (SpsWebSocketClient client : list) {
|
||||
client.close();
|
||||
}
|
||||
list.clear();
|
||||
list.add(this);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onOpen(ServerHandshake serverHandshake) {
|
||||
log.info("在线日志socket连接成功");
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onMessage(String s) {
|
||||
log.info("收到消息：" + s);
|
||||
//收到更新下载数据则赋值由外部处理，多个任务只处理一次避免重复处理
|
||||
if (s.contains(SocketMsgType.DL_ALL_DATA)) {
|
||||
this.excptMessage = s;
|
||||
} else {
|
||||
//通知类消息则需转发
|
||||
socketMsgService.sendNoticeMsg(s);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onClose(int i, String s, boolean b) {
|
||||
log.info("在线日志socket断开");
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onError(Exception e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onWebsocketPing(WebSocket conn, Framedata f) {
|
||||
try {
|
||||
Thread.sleep(1000 * 5);
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
send("---pong---");
|
||||
}
|
||||
|
||||
@Deprecated
|
||||
public static void destroy() {
|
||||
if (list.isEmpty())
|
||||
return;
|
||||
for (SpsWebSocketClient client : list) {
|
||||
client.close();
|
||||
}
|
||||
list.clear();
|
||||
|
||||
}
|
||||
|
||||
//发送消息
|
||||
public void sendMessage(String message) {
|
||||
this.send(message);
|
||||
System.out.println("已发送消息：" + message);
|
||||
}
|
||||
|
||||
//获取接收到的信息
|
||||
public String getExcptMessage() {
|
||||
if (excptMessage != null) {
|
||||
String message = new String(excptMessage);
|
||||
excptMessage = null;
|
||||
return message;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
@ -0,0 +1,118 @@
|
||||
package com.glxp.sale.admin.thread;
|
||||
|
||||
import com.glxp.sale.admin.constant.BasicExportTypeEnum;
|
||||
import com.glxp.sale.admin.constant.SocketMsgType;
|
||||
import com.glxp.sale.admin.entity.param.SystemParamConfigEntity;
|
||||
import com.glxp.sale.admin.entity.sync.SocketMsgEntity;
|
||||
import com.glxp.sale.admin.service.param.SystemParamConfigService;
|
||||
import com.glxp.sale.admin.socket.server.SpsSyncWebSocket;
|
||||
import com.glxp.sale.admin.util.RedisUtil;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import javax.annotation.Resource;
|
||||
import java.util.Arrays;
|
||||
|
||||
@Service
|
||||
public class HeartTaskService {
|
||||
|
||||
final Logger logger = LoggerFactory.getLogger(HeartTask.class);
|
||||
@Resource
|
||||
SystemParamConfigService systemParamConfigService;
|
||||
@Resource
|
||||
RedisUtil redisUtil;
|
||||
@Resource
|
||||
ScanUploadService scanUploadService;
|
||||
@Resource
|
||||
ScanDownloadService scanDownloadService;
|
||||
@Resource
|
||||
DlAllDataService dlAllDataService;
|
||||
@Resource
|
||||
SpsSyncWebSocket spsSyncWebSocket;
|
||||
@Value("${SPMS_WEBSOCKET_TOKEN}")
|
||||
private String socketToken;
|
||||
|
||||
//定时从上游下载数据
|
||||
public void dlData() {
|
||||
SystemParamConfigEntity upConnect = systemParamConfigService.selectByParamKey("sync_upstream_enable");
|
||||
if (upConnect != null && upConnect.getParamValue().equals("1")) {
|
||||
dlAllData();
|
||||
scanUpload();
|
||||
}
|
||||
SystemParamConfigEntity donwConnect = systemParamConfigService.selectByParamKey("sync_downstream_enable");
|
||||
if (donwConnect != null && donwConnect.getParamValue().equals("1")) {
|
||||
scanDonwload();
|
||||
}
|
||||
}
|
||||
|
||||
private void scanUpload() {
|
||||
SystemParamConfigEntity systemParamConfigEntity = systemParamConfigService.selectByParamKey("sc_udiinfo_upload");
|
||||
long timeInterval = Long.parseLong(systemParamConfigEntity.getParamValue()) * 60 * 1000;
|
||||
long curTime = System.currentTimeMillis();
|
||||
//定时扫描
|
||||
Long lastTime = (Long) redisUtil.get("UPLOAD_UDIINFO_STATUS");
|
||||
if (lastTime == null) {
|
||||
lastTime = System.currentTimeMillis();
|
||||
redisUtil.set("UPLOAD_UDIINFO_STATUS", lastTime, 30 * 60);
|
||||
}
|
||||
if (curTime - lastTime > timeInterval) {
|
||||
redisUtil.set("UPLOAD_UDIINFO_STATUS", curTime);
|
||||
scanUploadService.scanAllDatas();
|
||||
scanUploadService.scanAllBus();
|
||||
scanUploadService.scanAllOrders();
|
||||
scanUploadService.scanAllSchedule();
|
||||
}
|
||||
}
|
||||
|
||||
private void scanDonwload() {
|
||||
SystemParamConfigEntity upConnect = systemParamConfigService.selectByParamKey("sync_downstream_enable");
|
||||
if (upConnect.getParamValue().equals("1")) {
|
||||
SystemParamConfigEntity systemParamConfigEntity = systemParamConfigService.selectByParamKey("sc_udiinfo_status");
|
||||
long timeInterval = Long.parseLong(systemParamConfigEntity.getParamValue()) * 60 * 1000;
|
||||
long curTime = System.currentTimeMillis();
|
||||
Long lastTime = (Long) redisUtil.get("SC_UDIINFO_DOWNLOAD_STATUS");
|
||||
if (lastTime == null) {
|
||||
lastTime = System.currentTimeMillis();
|
||||
redisUtil.set("SC_UDIINFO_DOWNLOAD_STATUS", lastTime);
|
||||
}
|
||||
if (curTime - lastTime > timeInterval) {
|
||||
redisUtil.set("SC_UDIINFO_DOWNLOAD_STATUS", curTime);
|
||||
scanDownloadService.scanAllData();
|
||||
scanDownloadService.scanAllBus();
|
||||
scanDownloadService.scanAllOrder();
|
||||
scanDownloadService.scanScheduleList();
|
||||
scanDownloadService.scanUdis();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public void dlUpAllData() {
|
||||
SystemParamConfigEntity upConnect = systemParamConfigService.selectByParamKey("sync_upstream_enable");
|
||||
if (upConnect != null && upConnect.getParamValue().equals("1")) {
|
||||
dlAllData();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
public void dlAllData() {
|
||||
logger.info("定时从上游下载全部据-----");
|
||||
redisUtil.set("SC_UDIINFO_DOWNLOAD_STATUS", System.currentTimeMillis());
|
||||
String doing = (String) redisUtil.get("is_doing_download");
|
||||
if (doing == null || doing.equals("false")) {
|
||||
redisUtil.set("is_doing_download", "true", 60);
|
||||
dlAllDataService.dllNewAllOrder();
|
||||
Arrays.stream(BasicExportTypeEnum.values()).forEach(i -> {
|
||||
dlAllDataService.pullData(i);
|
||||
});
|
||||
dlAllDataService.dlAllDi();
|
||||
redisUtil.set("is_doing_download", "false");
|
||||
spsSyncWebSocket.sendMessage(SocketMsgEntity.builder().type(SocketMsgType.DL_ALL_DATA).content("").remark("下载基础信息").build(), "2:" + socketToken);
|
||||
|
||||
}
|
||||
logger.info("定时从上游下载全部据-----结束");
|
||||
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,48 @@
|
||||
package com.glxp.sale.admin.util;
|
||||
|
||||
import cn.hutool.core.util.ArrayUtil;
|
||||
import com.beust.jcommander.internal.Lists;
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
public class ByteArraySplitter {
|
||||
|
||||
/**
|
||||
* 对String分片转换为List<byte[]>
|
||||
*
|
||||
* @param source 字符串
|
||||
* @param size 分片的长度 单位字节
|
||||
* @return
|
||||
*/
|
||||
public static List<byte[]> split(String source, int size) {
|
||||
// 存放最终结果
|
||||
List<byte[]> result = Lists.newArrayList();
|
||||
|
||||
if (StringUtils.isEmpty(source)) {
|
||||
return null;
|
||||
}
|
||||
|
||||
byte[] sourceBytes = source.getBytes();
|
||||
if (size > sourceBytes.length) {
|
||||
result.add(sourceBytes);
|
||||
return result;
|
||||
}
|
||||
// 开始进行split
|
||||
int startIndex = 0;
|
||||
int endIndex = sourceBytes.length - 1;
|
||||
boolean isRunning = true;
|
||||
while (isRunning) {
|
||||
if ((endIndex + 1) - startIndex > size) {
|
||||
result.add(ArrayUtil.sub(sourceBytes, startIndex, startIndex + size));
|
||||
startIndex += size;
|
||||
} else {
|
||||
result.add(ArrayUtil.sub(sourceBytes, startIndex, endIndex + 1));
|
||||
isRunning = false;
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Loading…
Reference in New Issue