网络编程：使用udp实现数据文件的接收java

1、需求

2、逻辑实现

3、代码实现

4、总结

1、需求

发送方将文件按照数据帧进行发送，接收方完成数据接收的还原，即还原为相应的文件。

2、逻辑实现

采用ConcrrenutHashMap作为缓冲区，每次处理时都判断，数据是否连续，如果连续，就进行就根据数据偏移量完成数据文件的写入（数据偏移量是由帧头相应字段计算所得，是前期设计好的帧头），当达到缓冲区的某个阈值时，会对接收到的数据帧进行处理；如果前后接收到的数据帧时间超过某个阈值，就表示数据帧在传输过程中丢失了，那么就进行记录。

3、代码实现

package com.ruoyi.system.service.customService.method1;

/**
 * @Author 不要有情绪的  ljy
 * @Date 2024/5/17 11:35
 * @Description:
 */


import com.ruoyi.system.domain.NetworkConfig;
import com.ruoyi.system.service.INetworkConfigService;
import com.ruoyi.system.service.customService.dealGKService_NewThread.DealGkDataServiceSuperWithNewThread;
import com.ruoyi.system.service.customService.dealGKService_ThreadPool.DealGkDataServiceSuperWithThreadPool;
import com.ruoyi.system.service.customService.dealGKService_ThreadPool_Buffer.DealGkDataServiceSuperWithThreadPoolAndBuffer;
import com.ruoyi.system.service.customService.method2.SaveGKOriginalDataServiceWithBuffer;
import com.ruoyi.system.utlis.FileUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Service
public class UDPReceiverSuper {
    private static final int BUFFER_SIZE = 1044;
    private static final int HEAD_SIZE = 20;
    private static final int DATA_SIZE = 1024;
    private static final int MAX_BUFFER_SIZE = 1 * 1024 * 1024; // 缓冲器大小设置为1MB
    private static final double MAX_BUFFER_THRESHOLD = 0.8; // 缓冲区阈值
    private static final int MAX_BUFFER_INDEX = (int) (MAX_BUFFER_SIZE * MAX_BUFFER_THRESHOLD / DATA_SIZE); //缓冲区元素数量阈值

    //timestampToBufferMap存储的是：时间戳，TreeMap，TreeMap里面存储的是：当前包序号，接受数据的对象
    private Map<Long, ConcurrentHashMap<Long, DatagramPacket>> timestampToBufferMap = new HashMap();
    private long timeStamp;
    private boolean isClosed = false;// 使用阻塞队列作为缓冲区
    private long errorPackageSum = 0;
    private int frameNum;        //用于帧计数

    Thread udpReceiverThread;

    @Value("${GK.GKOriginalDataFilePath}")
    private String GKOriginalDataFilePath; // 管控原始数据文件存储路径

    @Value("${HP.storagePath}")
    private String storagePath;    //高性能数据接收路径
    @Autowired
    private INetworkConfigService networkConfigService;
    @Autowired
    private DealGkDataServiceSuperWithNewThread dealGkDataServiceSuperWithNewThread;
    @Autowired
    private DealGkDataServiceSuperWithThreadPoolAndBuffer dealGkDataServiceSuperWithThreadPoolAndBuffer;
    @Autowired
    private DealGkDataServiceSuperWithThreadPool dealGkDataServiceSuperWithThreadPool;
    @Autowired
    private SaveGKOriginalDataService saveGKOriginalDataService;
    @Autowired
    private SaveGKOriginalDataServiceWithBuffer saveGKOriginalDataServiceWithBuffer;



    public UDPReceiverSuper() {
    }

    public void start() {
        //创建父文件夹

        Path path = Paths.get(storagePath);
        if (Files.notExists(path)) {
            try {
                Files.createDirectories(path);
                System.out.println("Directories created successfully: " + storagePath);
            } catch (IOException e) {
                System.err.println("Failed to create directories: " + e.getMessage());
            }
        } else {
            System.out.println("Directories already exist: " + storagePath);
        }

        // 启动接收数据的线程
        if (udpReceiverThread == null) {
            udpReceiverThread = new Thread(new Receiver());
            udpReceiverThread.start();
        }
    }

    //数据帧头定义
    private class PackageHeader {
        public long id = 0;
        public long timestamp = 0;
        public long totalPackageNum = 0;
        public long currentPackageNum = 0;
        public long dataLength = 0;
    }


    // 接收数据的线程
    private class Receiver implements Runnable {
        @Override
        public void run() {
            NetworkConfig networkConfig = networkConfigService.selectNetworkConfigById(1L);
            String port = networkConfig.getPort();
            String ip = networkConfig.getIp();
            System.out.println("实际未绑定ip");
            System.out.println("ip: " + ip + "  port: " + port);
            try {
                DatagramSocket ds = new DatagramSocket(Integer.parseInt(port));
                if (ds != null) {
                    isClosed = false;
                }
                System.out.println("udpReceiver_ds: " + ds + "   等待接收数据......");


                while (true) {
                    if (isClosed) {
                        break;
                    }
                    byte[] receiveData = new byte[BUFFER_SIZE];   //接收数据缓存区，大小为1044
                    DatagramPacket receivePacket = new DatagramPacket(receiveData, receiveData.length);
                    ds.receive(receivePacket);     //接收数据
                    byte[] data1 = receivePacket.getData();
                    frameNum++;
//                    System.out.println("当前帧数为： " + frameNum);   //todo 用于打印输出当前接收到的帧数

                    ByteBuffer byteBuffer1 = ByteBuffer.allocate(data1.length);
                    byteBuffer1.put(data1);
                    byteBuffer1.flip();   //flip操作是将：写模式切换到读模式，将‘limit’设置为当前的‘position’，将‘position’重置为0
//                    ByteBuffer byteBuffer1 = ByteBuffer.allocate(receiveData.length);
//                    byteBuffer1.put(receiveData);
//                    byteBuffer1.flip();   //flip操作是将：写模式切换到读模式，将‘limit’设置为当前的‘position’，将‘position’重置为0
                    /*两种情况：1、接收管控  2、接收高性能*/
                    byteBuffer1.order(ByteOrder.LITTLE_ENDIAN);  //转化为小端
                    int headerType = byteBuffer1.getInt();       //得到设备标识符
                        //获取时间戳
                        byte[] data = receivePacket.getData();

                        //获取帧头信息
                        PackageHeader packageHeader = new PackageHeader();
                        for (int i = 0; i < 4; i++) {
                            packageHeader.id = (packageHeader.id << 8) + (data[i] & 0xFF);
                        }
                        for (int i = 4; i < 8; i++) {
                            packageHeader.timestamp = (packageHeader.timestamp << 8) + (data[i] & 0xFF);
                        }
                        for (int i = 8; i < 12; i++) {
                            packageHeader.totalPackageNum = (packageHeader.totalPackageNum << 8) + (data[i] & 0xFF);
                        }
                        for (int i = 12; i < 16; i++) {
                            packageHeader.currentPackageNum = (packageHeader.currentPackageNum << 8) + (data[i] & 0xFF);
                        }
                        for (int i = 16; i < 20; i++) {
                            packageHeader.dataLength = (packageHeader.dataLength << 8) + (data[i] & 0xFF);
                        }

                        //防止误码，判断当前时间戳是否在列表中（是否新启线程接收新的数据包）
                        /*
                         *   数据帧头合法性判定
                         *  1、设备ID小于20
                         *  2、时间戳事件小于1年
                         *  3、总数据帧数小于10000，小于100MB
                         *  4、当前数据帧数小于10000，小于100MB
                         *  5、有效数据长度小于等于1024
                         */
                        if (packageHeader.id < 20 &&
                                packageHeader.timestamp > 0 &&
                                packageHeader.timestamp < 60 * 60 * 24 * 365 &&
                                packageHeader.totalPackageNum < 100 * 1000 &&
                                packageHeader.currentPackageNum < 100 * 1000 &&
                                packageHeader.dataLength <= 1024
                        ) {

                            //数据接收进入缓冲区
                            if (!timestampToBufferMap.containsKey(packageHeader.timestamp)) {
                                long totalPackageNum = 0;
                                for (int i = 8; i < 12; i++) {
                                    totalPackageNum = (totalPackageNum << 8) + (data[i] & 0xFF);
                                }
                                if (totalPackageNum >= 100000) {   //防止误码
                                    continue;
                                }
                                timestampToBufferMap.put(packageHeader.timestamp, new ConcurrentHashMap<>());
                                new Thread(new MyRunnable(packageHeader.timestamp, totalPackageNum) {
                                }).start();
                            }


                            if (timestampToBufferMap.get(packageHeader.timestamp) != null) {
                                timestampToBufferMap.get(packageHeader.timestamp).put(packageHeader.currentPackageNum, receivePacket);  // 将接收到的数据包放入对应缓冲区
                            }
                        }else{
                            System.out.println("检测到误码数据。");
                        }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    // 带参数的Runnable实现类
    class MyRunnable implements Runnable {
        private Long stamp;
        private Long totalPackageNum;
        private Long currentPackageNum;
        private int times;

        // 构造函数，接收需要保存的值
        public MyRunnable(Long timestamp, Long totalPackageNum) {
            this.stamp = timestamp;
            this.totalPackageNum = totalPackageNum;
            this.currentPackageNum = new Long(0);
            this.times = 0;
        }

        @Override
        public void run() {
            //写磁盘文件
            RandomAccessFile raf = null;
            while (!(new File(storagePath + File.separator + stamp + ".cpio").exists()) || raf == null) {
                try {
                    raf = new RandomAccessFile(storagePath + File.separator + stamp + ".cpio", "rw");
                    Thread.sleep(1);
                } catch (FileNotFoundException | InterruptedException e) {
                    e.printStackTrace();
                }
            }

            // 在线程中使用保存的值
            try {
                long lastReceivedTime = 0;
                long receivedPackSum = 0;
                long fileSize = totalPackageNum * 1024;  // 只有首次创建完cpio包后，设置cpio包的大小
                long lastMapSize = 0;
                Map<Long, DatagramPacket> bufferMap = timestampToBufferMap.get(stamp);
                raf.setLength(fileSize);  // 设置cpio包的大小
                raf.getFD().sync();

                while (true) {
                    if (lastMapSize != bufferMap.size()) { //如果缓冲区中有数据写入
                        lastMapSize = bufferMap.size();
                        System.out.println(stamp + ".cpio文件已缓存数据：" + (times * MAX_BUFFER_INDEX + lastMapSize) + "/" + totalPackageNum + "帧。");
                        lastReceivedTime = System.currentTimeMillis();
                    } else {
                        long currentTime = System.currentTimeMillis();
                        //如果超过规定时间未接收到当前数据包中的数据帧,或者接收的数据帧数量大于等于总包数，或者缓冲区的数据量已经大于等于总包数
                        if (bufferMap.size() >= totalPackageNum || receivedPackSum >= totalPackageNum || currentTime - lastReceivedTime > 10000) {
                            if ((times * MAX_BUFFER_INDEX + lastMapSize) != totalPackageNum) {     //判断未完成接收的cpio包数量
                                System.out.println(stamp + ".cpio文件未完整接收，当前未完整接收的cpio包的总数量:" + ++errorPackageSum + "。");
                            }
                            if (bufferMap.size() > 0) {
                                times++;
                                currentPackageNum = persistence(stamp, bufferMap, raf, currentPackageNum);
                            }
                            timestampToBufferMap.remove(stamp);  //移除缓冲区
                            raf.close();

                            Thread.sleep(1000);
                            new Thread(new Runnable() {
                                @Override
                                public void run() {
                                    //解压cpio包
                                    //将cpio包解压，得到所有的文件名，并存储到数据库
                                    try {
                                        String cpioFilePath = storagePath + File.separator + stamp + ".cpio";
                                        String outputPath = storagePath;
                                        boolean isUnzipSuccess = FileUtil.unCpioFile(cpioFilePath, outputPath);
                                        if (isUnzipSuccess) {
                                            System.out.println("解压成功！");
                                        }
                                    } catch (IOException e) {
                                        e.printStackTrace();
                                    }
                                }
                            }).start();
                            return; //结束线程
                        }
                    }

                    // 缓冲区大小达到阈值时，进行缓冲区数据的处理
                    if (bufferMap.size() > MAX_BUFFER_INDEX) {
                        //计算已接收的帧数
                        times++;
                        receivedPackSum += bufferMap.size();
                        currentPackageNum = persistence(stamp, bufferMap, raf, currentPackageNum);
                    }
                    Thread.sleep(100);  //单次延时时间100ms,期间至少缓存250k数据，和缓冲区要匹配
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        /**
         * 持久化
         *
         * @param bufferMap
         * @param raf
         * @param currentPackageNum
         * @return
         */
        private Long persistence(Long stamp, Map<Long, DatagramPacket> bufferMap, RandomAccessFile raf, Long currentPackageNum) {
            try {
                long firstPackageNum = -1; //记录连续的第一个包序号
                List<byte[]> dataList = new ArrayList<>();
                long thisTimeSum = 0;    //此次写入的数据量
                long searchIndex = 0;    //搜索偏移量
                long lastUnsuccessFindMinIndex = currentPackageNum + MAX_BUFFER_INDEX; //本次未成功写入的最小下标
                //忽略数据连续性，保证写入MAX_BUFFER_INDEX个数据帧
                //之所以要引入bufferMap.size() > 0的判定是用于处理“接收超时时缓存区中的数据帧”和“最后一次缓存区不满时的数据帧”
                while (bufferMap.size() > 0 && thisTimeSum < MAX_BUFFER_INDEX) {
                    while (bufferMap.containsKey(currentPackageNum + searchIndex)) {
                        if (firstPackageNum == -1) {
                            firstPackageNum = currentPackageNum + searchIndex;
                        }
                        DatagramPacket datagramPacket = bufferMap.get(currentPackageNum + searchIndex);  //读取数据包
                        bufferMap.remove(currentPackageNum + searchIndex++);
                        int dataLength = datagramPacket.getLength() - HEAD_SIZE;    //获取本次的数据包长度
                        byte[] byteBuffer = new byte[dataLength];   //临时数组
                        System.arraycopy(datagramPacket.getData(), HEAD_SIZE, byteBuffer, 0, dataLength);  //数据存入临时数组
                        dataList.add(byteBuffer);   //加入到数据列表
                        if (++thisTimeSum >= MAX_BUFFER_INDEX) {  //连续数据大于单词写入数据限制
                            break;
                        }
                    }

                    //记录未成功写入的最小下标
                    if (lastUnsuccessFindMinIndex == currentPackageNum + MAX_BUFFER_INDEX) {
                        lastUnsuccessFindMinIndex = currentPackageNum + searchIndex;
                    }

                    //持久化
                    if (dataList.size() > 0) {
                        // 计算总长度
                        int totalLength = 0;
                        for (byte[] array : dataList) {
                            totalLength += array.length;
                        }

                        // 创建目标数组
                        byte[] data = new byte[totalLength];
                        int currentPosition = 0;

                        // 复制数据
                        for (byte[] array : dataList) {
                            System.arraycopy(array, 0, data, currentPosition, array.length);
                            currentPosition += array.length;
                        }

                        long offset = firstPackageNum * DATA_SIZE;   //根据当前包序号，计算写入偏移量
                        raf.seek(offset);   //在偏移量后，写入数据
                        raf.write(data);    //写入数据

                        System.out.println(stamp + ".cpio文件，已完成第" + times + "次数据的硬盘写入操作，单次写入数据大小：" + (totalLength / 1024) + "kB。");
                        if (lastUnsuccessFindMinIndex == currentPackageNum + MAX_BUFFER_INDEX || lastUnsuccessFindMinIndex == totalPackageNum) {
                            System.out.println("写入至此，未检测到或已修复丢失数据帧（丢失/乱序到达）。");
                        } else {
                            System.out.println("写入至此，已检测到且仍存在丢失数据帧（丢失/乱序到达）。");
                        }
                    }

                    dataList.clear();   //清空连续数据
                    searchIndex++;  //下标+1重新搜索
                }
                return lastUnsuccessFindMinIndex;  //将未成功的最小下标返回
            } catch (IOException e) {
                e.printStackTrace();
            }
            return currentPackageNum;
        }
    }


}