增加socket通讯，增加移动端控制接口

bsw215583320

2023-09-12 a6ccb31191385dd9c3d2e9d75e1bcda7ff6a4ed3

增加socket通讯，增加移动端控制接口

 jeecg-module-dry/jeecg-module-dry-api/src/main/java/org/jeecg/modules/dry/common/CacheConstants.java

@@ -1,8 +1,19 @@
package org.jeecg.modules.dry.common;

import org.jeecg.modules.dry.vo.DryFault;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public interface CacheConstants {
    enum RedisKeyEnum{
        WORK_ORDER("workOrder","工单MAP");
        WORK_ORDER("workOrder","工单MAP"),


        EQP_MAP("eqpMap", "设备MAP,key:tenantId+equipmentId");

        private String code;
        private String text;
        RedisKeyEnum(String code, String text) {
@@ -18,4 +29,6 @@
            return text;
        }
    }


}

 jeecg-module-dry/jeecg-module-dry-api/src/main/java/org/jeecg/modules/dry/common/CommonDict.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,9 @@
package org.jeecg.modules.dry.common;

import java.util.HashMap;
import java.util.Map;

public class CommonDict {

    public static Map<String, String> faultDict = new HashMap<>();
}

 jeecg-module-dry/jeecg-module-dry-api/src/main/java/org/jeecg/modules/dry/entity/DryEquipment.java

@@ -10,6 +10,7 @@
import com.baomidou.mybatisplus.annotation.TableLogic;
import lombok.Data;
import com.fasterxml.jackson.annotation.JsonFormat;
import lombok.ToString;
import org.jeecg.modules.dry.vo.RealTimeDataVo;
import org.springframework.format.annotation.DateTimeFormat;
import org.jeecgframework.poi.excel.annotation.Excel;
@@ -29,6 +30,7 @@
@TableName("dry_equipment")
@Accessors(chain = true)
@EqualsAndHashCode(callSuper = false)
@ToString
@ApiModel(value="dry_equipment对象", description="干燥机")
public class DryEquipment implements Serializable {
    private static final long serialVersionUID = 1L;
@@ -84,6 +86,11 @@
    @ApiModelProperty(value = "租户id")
    private Integer tenantId;

    /**设备IP*/
    @Excel(name = "设备IP", width = 15)
    @ApiModelProperty(value = "设备IP")
    private String ip;

    public DryEquipment() {
    }
    public DryEquipment(RealTimeDataVo realTimeDataVo) {

 jeecg-module-dry/jeecg-module-dry-api/src/main/java/org/jeecg/modules/dry/vo/CommandMessageVo.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,26 @@
package org.jeecg.modules.dry.vo;

import lombok.Data;
import lombok.ToString;

@Data
@ToString
public class CommandMessageVo {
    /**租户ID*/
    private Integer tenantId;
    /**设备编号*/
    private String machineId;

    /**指令编码**/
    private String code;
    /**消息内容**/
    private String msg;

    /**
     * 1001 é£Žç®±å‡     1002 é£Žç®±é™
     * 1003 æ»šç­’é•¿     1004 æ»šç­’降
     * 1005 æ»šç­’正转    1006 æ»šç­’反转
     * 1007 è®¾å¤‡åœæ­¢
     */

}

 jeecg-module-dry/jeecg-module-dry-api/src/main/java/org/jeecg/modules/dry/vo/DryFault.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,10 @@
package org.jeecg.modules.dry.vo;

import lombok.Data;

@Data
public class DryFault {

    private String code;
    private String text;
}

 jeecg-module-dry/jeecg-module-dry-api/src/main/java/org/jeecg/modules/dry/vo/DryOrderVo.java

@@ -115,6 +115,8 @@
    @ApiModelProperty(value = "故障")
    private String fault;

    private List<String> faultList = new ArrayList<>();

    @ApiModelProperty(value = "明细数据列表")
    private List<DryOrderTrendVo> detailList = new ArrayList<>();
    @ApiModelProperty(value = "明细")

 jeecg-module-dry/jeecg-module-dry-api/src/main/java/org/jeecg/modules/dry/vo/SocketMsgVo.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,23 @@
package org.jeecg.modules.dry.vo;


import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.ToString;

import java.io.Serializable;

@NoArgsConstructor
@Data
@ToString
public class SocketMsgVo implements Serializable {
    /**指令编码**/
    private String code;
    /**消息内容**/
    private String msg;

    public SocketMsgVo(CommandMessageVo vo) {
        this.code = vo.getCode();
        this.msg = vo.getMsg();
    }
}

 jeecg-module-dry/jeecg-module-dry-biz/pom.xml

@@ -14,6 +14,12 @@
            <artifactId>jeecg-module-dry-api</artifactId>
            <version>3.5.0</version>
        </dependency>

        <dependency>
            <groupId>org.apache.mina</groupId>
            <artifactId>mina-core</artifactId>
            <version>2.0.17</version>
        </dependency>
    </dependencies>
    <build>
        <!-- æ‰“包名称 -->

 jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/controller/DryEquipmentController.java

@@ -104,6 +104,10 @@
        return Result.OK(eqps);
    }





    /**
     *   æ·»åŠ 
     *

 jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/controller/DryRealTimeDataController.java

@@ -8,6 +8,7 @@

import org.jeecg.modules.dry.service.*;

import org.jeecg.modules.dry.vo.CommandMessageVo;
import org.jeecg.modules.dry.vo.RealTimeDataVo;

import org.springframework.beans.factory.annotation.Autowired;
@@ -44,4 +45,17 @@
        return dryRealTimeDataService.queryMachineRealTImeData(realTimeDataVo);
    }


    /**
     * 1001 é£Žç®±å‡     1002 é£Žç®±é™
     * 1003 æ»šç­’升     1004 æ»šç­’降
     * 1005 æ»šç­’正转    1006 æ»šç­’反转
     * 1007 è®¾å¤‡åœæ­¢
     */
    @ApiOperation(value="发送控制指令", notes="向服务端发送控制指令，由服务端通过socket转发给控制模块")
    @PostMapping("/sendCommand")
    public Result<?> sendCommand(@RequestBody CommandMessageVo msgVo) {
        return dryRealTimeDataService.sendSocketMsg(msgVo);
    }

}

 jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/listener/InitListener.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,34 @@
package org.jeecg.modules.dry.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.google.common.io.ByteStreams;
import org.jeecg.modules.dry.common.CacheConstants;
import org.jeecg.modules.dry.common.CommonDict;
import org.jeecg.modules.dry.vo.DryFault;
import org.springframework.boot.context.event.ApplicationStartedEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.io.InputStream;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Component
public class InitListener implements ApplicationListener<ApplicationStartedEvent> {
    @Override
    public void onApplicationEvent(ApplicationStartedEvent event) {
        InputStream fault = getClass().getResourceAsStream("/fault.json");
        try {
            JSONObject jsonObject = JSONObject.parseObject(fault, JSONObject.class);
            Map<String, String> map = jsonObject.toJavaObject(Map.class);
            CommonDict.faultDict = map;

        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}

 jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/service/IDryEquipmentService.java

@@ -3,6 +3,9 @@
import org.jeecg.modules.dry.entity.DryEquipment;
import com.baomidou.mybatisplus.extension.service.IService;

import java.util.HashMap;
import java.util.List;

/**
 * @Description: å¹²ç‡¥æœº
 * @Author: jeecg-boot
@@ -11,4 +14,6 @@
 */
public interface IDryEquipmentService extends IService<DryEquipment> {

    DryEquipment selectByTenantIdEquipmentId(String tenantId, String equipmentId);

}

 jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/service/IDryRealTimeDataService.java

@@ -1,10 +1,13 @@
package org.jeecg.modules.dry.service;

import org.jeecg.common.api.vo.Result;
import org.jeecg.modules.dry.vo.CommandMessageVo;
import org.jeecg.modules.dry.vo.RealTimeDataVo;

public interface IDryRealTimeDataService {
    Result<?> realTimeDataHandle(RealTimeDataVo realTimeDataVo);

    Result<?> queryMachineRealTImeData(RealTimeDataVo realTimeDataVo);

    Result<?> sendSocketMsg(CommandMessageVo msgVo);
}

 jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/service/impl/DryEquipmentServiceImpl.java

@@ -1,11 +1,17 @@
package org.jeecg.modules.dry.service.impl;

import org.jeecg.common.util.RedisUtil;
import org.jeecg.modules.dry.common.CacheConstants;
import org.jeecg.modules.dry.entity.DryEquipment;
import org.jeecg.modules.dry.mapper.DryEquipmentMapper;
import org.jeecg.modules.dry.service.IDryEquipmentService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;

import java.util.HashMap;
import java.util.List;

/**
 * @Description: å¹²ç‡¥æœº
@@ -16,4 +22,20 @@
@Service
public class DryEquipmentServiceImpl extends ServiceImpl<DryEquipmentMapper, DryEquipment> implements IDryEquipmentService {

    @Autowired
    private RedisUtil redisUtil;
    @Override
    public DryEquipment selectByTenantIdEquipmentId(String tenantId, String equipmentId) {
        DryEquipment dryEquipment = (DryEquipment) redisUtil.hget(CacheConstants.RedisKeyEnum.EQP_MAP.getCode(), tenantId + equipmentId);
        if (dryEquipment == null) {
            List<DryEquipment> list = this.list();
            for (DryEquipment equipment : list) {
                redisUtil.hset(CacheConstants.RedisKeyEnum.EQP_MAP.getCode(),equipment.getTenantId()+equipment.getCode(),equipment);
                if ((equipment.getTenantId()+equipment.getCode()).equals(tenantId + equipmentId)) {
                    dryEquipment = equipment;
                }
            }
        }
        return dryEquipment;
    }
}

 jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/service/impl/DryRealTimeDataServiceImpl.java

@@ -5,24 +5,32 @@
import com.alibaba.fastjson.TypeReference;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.mina.core.service.IoAcceptor;
import org.apache.mina.core.service.IoHandler;
import org.apache.mina.core.session.IoSession;
import org.jeecg.common.api.vo.Result;
import org.jeecg.common.config.TenantContext;
import org.jeecg.common.util.RedisUtil;
import org.jeecg.modules.dry.common.CacheConstants;
import org.jeecg.modules.dry.entity.*;
import org.jeecg.modules.dry.service.*;
import org.jeecg.modules.dry.vo.DryOrderTrendVo;
import org.jeecg.modules.dry.vo.DryOrderVo;
import org.jeecg.modules.dry.vo.RealTimeDataVo;
import org.jeecg.modules.dry.socket.ServerHandler;
import org.jeecg.modules.dry.socket.SocketServerConfig;
import org.jeecg.modules.dry.vo.*;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.io.IOException;
import java.io.ObjectOutputStream;
import java.net.Socket;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;

@Slf4j
@Service
public class DryRealTimeDataServiceImpl implements IDryRealTimeDataService {
@@ -42,6 +50,9 @@

    @Autowired
    private IDryProdRecordService prodRecordService;

    @Autowired
    private IoAcceptor ioAcceptor;

    @Override
    @Transactional
@@ -88,6 +99,13 @@
        orderVo.setState_windbox(realTimeDataVo.getState_windbox());
        orderVo.setState_valve(realTimeDataVo.getState_valve());
        orderVo.setOrderStatus(realTimeDataVo.getWorkorder_status());

        String fault = realTimeDataVo.getFault();
//        String[] split = fault.split(",");
//        for (String s : split) {
//
//        }

        DryOrderTrendVo trendVo = new DryOrderTrendVo(realTimeDataVo);
        // 2.2 ä¿å­˜å·¥å•å«æ°´çŽ‡å˜åŒ– æˆ– é‡é‡å˜åŒ–

@@ -292,7 +310,7 @@
            if (order != null) {
                double v = order.getOriginWeight() - order.getYield();
                list.add(item.getName().substring(0, item.getName().indexOf('#')+1));
                if (v > 0) {
                            if (v > 0 && order.getDryTime()>0) {
                    DecimalFormat df = new DecimalFormat("#.00");
                    dList.add(Double.valueOf(df.format(v / order.getDryTime() * 60)));
                } else {
@@ -311,4 +329,36 @@
}
        return Result.ok(orderVo);
    }


    @Override
    public Result<?> sendSocketMsg(CommandMessageVo msgVo) {


        DryEquipment dryEquipment = equipmentService.selectByTenantIdEquipmentId(msgVo.getTenantId() + "", msgVo.getMachineId());
        log.info("获取设备：" + dryEquipment.toString());

       // managedSessions.keySet().forEach(addr -> {
           // ObjectOutputStream oos = null;
            try {
//                Socket socket = SocketServerConfig.clientMap.get(addr);
                IoSession session = ServerHandler.clientSocket.get(dryEquipment.getIp());
                if (session == null) {
                    return Result.error("未获取到session,请检查客户端配置或设备ip配置是否正常");
                }
                SocketMsgVo smv = new SocketMsgVo(msgVo);
                session.write(JSONObject.toJSONString(smv));
//                oos = new ObjectOutputStream(socket.getOutputStream());
//                String s = JSONObject.toJSONString(new SocketMsgVo(msgVo));
//                oos.writeUTF(s);
//                oos.flush();

            } catch (Exception e) {
                throw new RuntimeException(e);
            } finally {

            }
     //   });
        return Result.OK();
    }
}

 jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/socket/MinaConfig.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,59 @@
package org.jeecg.modules.dry.socket;

import lombok.extern.slf4j.Slf4j;
import org.apache.mina.core.service.IoAcceptor;
import org.apache.mina.core.service.IoHandler;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.textline.TextLineCodecFactory;

import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


import java.net.InetSocketAddress;
import java.nio.charset.Charset;

@Slf4j
@Configuration
public class MinaConfig {
    // socket占用端口
    @Value("${mina.port}") 
    private int port;

    /** 15秒发送一次心跳包 */
    private static final int HEARTBEATRATE = 15;

    @Bean
    public LoggingFilter loggingFilter() {
        return new LoggingFilter();
    }

    @Bean
    public IoHandler ioHandler() { return new ServerHandler(); }

    @Bean
    public InetSocketAddress inetSocketAddress() {
        return new InetSocketAddress(port);
    }

    @Bean
    public IoAcceptor ioAcceptor() throws Exception {
        IoAcceptor acceptor = new NioSocketAcceptor();
        acceptor.getFilterChain().addLast("logger", loggingFilter());
        acceptor.getFilterChain().addLast("coderc",   // ä½¿ç”¨è‡ªå®šä¹‰ç¼–码解码工厂类
                new ProtocolCodecFilter(new TextLineCodecFactory(Charset.forName("UTF-8"))));//设置编码过滤器
        acceptor.getSessionConfig().setReadBufferSize(1024*1024);//设置缓冲区
        acceptor.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, 10);  //配置会话信息

        acceptor.setHandler(ioHandler()); //自定义处理业务的代码：自定义的类
        acceptor.bind(inetSocketAddress());//绑定端口号
        log.info("Socket服务器在端口：" + port + "已经启动");

        return acceptor;
    }
}


 jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/socket/ServerHandler.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,78 @@
package org.jeecg.modules.dry.socket;


import lombok.extern.slf4j.Slf4j;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.concurrent.ConcurrentHashMap;


@Slf4j
public class ServerHandler extends IoHandlerAdapter {

    public static ConcurrentHashMap<String, IoSession> clientSocket = new ConcurrentHashMap<>();

    @Override
    public void exceptionCaught(IoSession session, Throwable cause) throws Exception {
        log.error("出现异常 :" + session.getRemoteAddress().toString() + " : " + cause.toString());
        session.closeNow();

    }


    @Override
    public void sessionCreated(IoSession session) throws Exception {
        log.info("连接创建 : " + session.getRemoteAddress().toString());
        String remoteAddr = session.getRemoteAddress().toString().replace("/", "");
        String remoteIp = remoteAddr.substring(0, remoteAddr.indexOf(":"));
        clientSocket.put(remoteIp, session);
        session.getConfig().setIdleTime(IdleStatus.BOTH_IDLE, 10);
    }

    @Override
    public void sessionOpened(IoSession session) throws Exception {
        System.out.println(session);
        log.info("连接打开: " + session.getRemoteAddress().toString());
    }

    @Override
    public void messageReceived(IoSession session, Object message) throws Exception {
        log.info("接受到数据 :" + message);


    }


    @Override
    public void messageSent(IoSession session, Object message) throws Exception {
        log.info("返回消息内容 : " + message.toString());
    }

    @Override
    public void sessionIdle(IoSession session, IdleStatus status) throws Exception {
        if (status == IdleStatus.READER_IDLE) {
            //log.info("进入读空闲状态");
            //session.closeNow();
        } else if (status == IdleStatus.BOTH_IDLE) {
          //  log.info("BOTH空闲");
            //session.closeNow();
        }
    }

    @Override
    public void sessionClosed(IoSession session) throws Exception {
        String address = session.getRemoteAddress().toString().replace("/", "");
        log.info(address + "  ä¼šè¯å…³é—­äº†!");
        String remoteAddr = session.getRemoteAddress().toString().replace("/", "");
        String remoteIp = remoteAddr.substring(0, remoteAddr.indexOf(":"));
        clientSocket.remove(remoteIp);

        }

}



 jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/socket/ServerReceiveThread.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,53 @@
package org.jeecg.modules.dry.socket;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;

import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.Socket;

@Slf4j
public class ServerReceiveThread implements Runnable{
    private Socket socket;

    public ServerReceiveThread(Socket socket) {
        this.socket = socket;
    }

    @Override
    public void run() {
        try {
            //输入流接收数据
            ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
            //输出流发送数据
            ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());

            while (true) {

                String o = (String) ois.readUTF();
                JSONObject jsonObject = JSONObject.parseObject(o);
                System.out.println(jsonObject.toJSONString());
                String message = jsonObject.getString("msg");
                if ("close".equals(message)) {
                    oos.writeUTF("close");
                    oos.flush();
                    break;
                } else {
                    oos.writeUTF("接收数据成功:" + message);
                    oos.flush();
                }
            }
            log.info("服务端关闭客户端[{}]", socket.getRemoteSocketAddress());
            oos.close();
            ois.close();
            socket.close();
        } catch (Exception e) {
            log.info("接收数据异常socket关闭");
            e.printStackTrace();
        } finally {
            log.info("数据异常数据要怎么保留");
        }
    }

}

 jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/socket/SocketServerConfig.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,47 @@
package org.jeecg.modules.dry.socket;

import lombok.extern.slf4j.Slf4j;
import org.checkerframework.checker.units.qual.C;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Slf4j
public class SocketServerConfig {
    public static ServerSocket serverSocket = null;

    public static ConcurrentHashMap<String, Socket> clientMap = new ConcurrentHashMap<>();


    private static final ThreadPoolExecutor threadpool = new ThreadPoolExecutor(15, 15,
            10L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());

    public void createSocket() {

            new Thread(() -> {
                try {
                    serverSocket = new ServerSocket(1115);
                    log.info("开启socket监听，端口：1115");
                    while (true) {
                        Socket socket = serverSocket.accept();
                        clientMap.put(socket.getRemoteSocketAddress().toString(), socket);
                        log.info("接收到客户端连接:" + socket.getRemoteSocketAddress());
                        threadpool.execute(new ServerReceiveThread(socket));
                    }
                }catch (IOException e) {
                    log.error("socket服务启动异常");
                    e.printStackTrace();
                }
            }).start();

    }
}

 jeecg-module-dry/jeecg-module-dry-start/src/main/resources/fault.json

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,4 @@
{
  "6.1": "急停故障",
  "6.2": "其它故障"
}