新增websocket通信（中继服务同步相关）
anthonywj
2 years ago
parent
6647fbb9d7
commit
1476a0f894
@ -0,0 +1,16 @@
|
|||||||
|
package com.glxp.api.constant;
|
||||||
|
|
||||||
|
public interface SocketMsgType {
|
||||||
|
|
||||||
|
|
||||||
|
String DL_ALL_DATA = "DL_ALL_DATA"; //生产入库
|
||||||
|
String DL_NOTICE = "DL_NOTICE"; //通知类消息
|
||||||
|
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 任务类型
|
||||||
|
*/
|
||||||
|
|
||||||
|
String TASK_INVOICE_CONFIRM = "TASK_INVOICE_CONFIRM"; //发票确认
|
||||||
|
|
||||||
|
}
|
||||||
@ -0,0 +1,20 @@
|
|||||||
|
package com.glxp.api.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,40 @@
|
|||||||
|
package com.glxp.api.service.sync;
|
||||||
|
|
||||||
|
import cn.hutool.core.util.StrUtil;
|
||||||
|
import com.alibaba.fastjson.JSONObject;
|
||||||
|
import com.alibaba.fastjson.TypeReference;
|
||||||
|
import com.glxp.api.common.res.BaseResponse;
|
||||||
|
import com.glxp.api.constant.SocketMsgType;
|
||||||
|
import com.glxp.api.entity.sync.SocketMsgEntity;
|
||||||
|
import org.springframework.beans.factory.annotation.Value;
|
||||||
|
import org.springframework.scheduling.annotation.Async;
|
||||||
|
import org.springframework.stereotype.Service;
|
||||||
|
|
||||||
|
import javax.annotation.Resource;
|
||||||
|
|
||||||
|
@Service
|
||||||
|
public class SocketMsgService {
|
||||||
|
|
||||||
|
@Value("${SPMS_WEBSOCKET_TOKEN}")
|
||||||
|
String token;
|
||||||
|
|
||||||
|
@Async
|
||||||
|
public void dealNoticeMsg(SocketMsgEntity socketMsgEntity) {
|
||||||
|
if (socketMsgEntity != null && StrUtil.isNotEmpty(socketMsgEntity.getType())) {
|
||||||
|
switch (socketMsgEntity.getType()) {
|
||||||
|
case SocketMsgType.TASK_INVOICE_CONFIRM:
|
||||||
|
invoiceConfirm(socketMsgEntity);
|
||||||
|
break;
|
||||||
|
default:
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
//发票确认
|
||||||
|
public void invoiceConfirm(SocketMsgEntity socketMsgEntity) {
|
||||||
|
|
||||||
|
}
|
||||||
|
}
|
||||||
@ -0,0 +1,120 @@
|
|||||||
|
package com.glxp.api.service.sync;
|
||||||
|
|
||||||
|
import cn.hutool.core.util.StrUtil;
|
||||||
|
import com.alibaba.fastjson.JSONObject;
|
||||||
|
import com.glxp.api.constant.SocketMsgType;
|
||||||
|
import com.glxp.api.entity.sync.SocketMsgEntity;
|
||||||
|
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<>();
|
||||||
|
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 (StrUtil.isNotEmpty(s)) {
|
||||||
|
SocketMsgEntity socketMsgEntity = JSONObject.parseObject(s, SocketMsgEntity.class);
|
||||||
|
//收到更新下载数据则赋值由外部处理，多个任务只处理一次避免重复处理
|
||||||
|
if (socketMsgEntity.getType().equals(SocketMsgType.DL_ALL_DATA)) {
|
||||||
|
this.excptMessage = s;
|
||||||
|
} else {
|
||||||
|
//通知类消息则需转发
|
||||||
|
socketMsgService.dealNoticeMsg(socketMsgEntity);
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
log.error("无法识别消息！");
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
@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,105 @@
|
|||||||
|
package com.glxp.api.task;
|
||||||
|
|
||||||
|
import com.glxp.api.constant.BasicExportTypeEnum;
|
||||||
|
import com.glxp.api.dao.system.SyncDataSetDao;
|
||||||
|
import com.glxp.api.entity.system.SyncDataSetEntity;
|
||||||
|
import com.glxp.api.service.sync.HeartService;
|
||||||
|
import com.glxp.api.util.RedisUtil;
|
||||||
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.apache.commons.lang3.exception.ExceptionUtils;
|
||||||
|
import org.springframework.stereotype.Service;
|
||||||
|
|
||||||
|
import javax.annotation.Resource;
|
||||||
|
import java.util.Arrays;
|
||||||
|
|
||||||
|
@Slf4j
|
||||||
|
@Service
|
||||||
|
public class SyncHeartService {
|
||||||
|
|
||||||
|
@Resource
|
||||||
|
RedisUtil redisUtil;
|
||||||
|
@Resource
|
||||||
|
HeartService heartService;
|
||||||
|
@Resource
|
||||||
|
private SyncDataSetDao syncDataSetDao;
|
||||||
|
|
||||||
|
public void syncProcess() {
|
||||||
|
//查询数据同步设置
|
||||||
|
pushData();
|
||||||
|
pushOrder();
|
||||||
|
pullData();
|
||||||
|
}
|
||||||
|
|
||||||
|
public void pushData() {
|
||||||
|
SyncDataSetEntity syncDataSetEntity = syncDataSetDao.selectSet();
|
||||||
|
if (!syncDataSetEntity.isDownstreamEnable()) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
//定时上传最近更新基础数据至上游轮询时间
|
||||||
|
long timeInterval1 = syncDataSetEntity.getSyncTime() * 6 * 1000L;
|
||||||
|
long curTime1 = System.currentTimeMillis();
|
||||||
|
Long lastTime1 = (Long) redisUtil.get("SPS_SYNC_UPLOAD_DATA");
|
||||||
|
if (lastTime1 == null) {
|
||||||
|
lastTime1 = System.currentTimeMillis();
|
||||||
|
redisUtil.set("SPS_SYNC_UPLOAD_DATA", lastTime1);
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
if (curTime1 - lastTime1 > timeInterval1) {
|
||||||
|
Arrays.stream(BasicExportTypeEnum.values()).forEach(i -> {
|
||||||
|
heartService.pushData(syncDataSetEntity, null, i);
|
||||||
|
});
|
||||||
|
redisUtil.set("SPS_SYNC_UPLOAD_DATA", curTime1);
|
||||||
|
}
|
||||||
|
} catch (Exception e) {
|
||||||
|
log.error(ExceptionUtils.getStackTrace(e));
|
||||||
|
e.printStackTrace();
|
||||||
|
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
public void pushOrder() {
|
||||||
|
SyncDataSetEntity syncDataSetEntity = syncDataSetDao.selectSet();
|
||||||
|
if (!syncDataSetEntity.isDownstreamEnable()) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
//定时上传最近更新单据数据至上游轮询时间
|
||||||
|
long timeInterval2 = syncDataSetEntity.getOrderSyncTime() * 6 * 1000L;
|
||||||
|
long curTime2 = System.currentTimeMillis();
|
||||||
|
Long lastTime2 = (Long) redisUtil.get("SPS_SYNC_UPLOAD_ORDER");
|
||||||
|
if (lastTime2 == null) {
|
||||||
|
lastTime2 = System.currentTimeMillis();
|
||||||
|
redisUtil.set("SPS_SYNC_UPLOAD_ORDER", lastTime2);
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
if (curTime2 - lastTime2 > timeInterval2) {
|
||||||
|
heartService.uploadAllOrder(null);
|
||||||
|
heartService.uploadAllBusOrder(null);
|
||||||
|
redisUtil.set("SPS_SYNC_UPLOAD_ORDER", curTime2);
|
||||||
|
}
|
||||||
|
} catch (Exception e) {
|
||||||
|
log.error(ExceptionUtils.getStackTrace(e));
|
||||||
|
e.printStackTrace();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
public void pullData() {
|
||||||
|
try {
|
||||||
|
heartService.dlAllOrder();
|
||||||
|
} catch (Exception e) {
|
||||||
|
e.printStackTrace();
|
||||||
|
log.error(ExceptionUtils.getStackTrace(e));
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
heartService.dlAllDiProducts();
|
||||||
|
} catch (Exception e) {
|
||||||
|
e.printStackTrace();
|
||||||
|
log.error(ExceptionUtils.getStackTrace(e));
|
||||||
|
}
|
||||||
|
Arrays.stream(BasicExportTypeEnum.values()).forEach(i -> {
|
||||||
|
heartService.pullData(i);
|
||||||
|
});
|
||||||
|
redisUtil.set("SPS_SYNC_DOWNLOAD_DATA", System.currentTimeMillis());
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
Loading…
Reference in New Issue