[网络协议]UDP实现聊天功能

package com.han.network.udp;

import jdk.jfr.Description;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.net.*;
import java.nio.charset.StandardCharsets;
import java.text.DateFormat;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Scanner;
import java.util.Vector;

/**
 * Created with IntelliJ IDEA.
 *
 * @Date: 2022/4/7.
 * @Author: H.
 * @Description: 使用UDP实现了聊天功能，多线程进行通信，接收和发送功能有各自的线程，
 * 没有考虑线程死锁和通信问题，使用了一个临时缓冲解决了UDP每次只能传输64kb数据，
 * 理论上，上限取决于这个临时缓冲的大小
 */
public class UDPUtils implements Runnable {

    private int waitPort; // 设置UDP等待接收消息的端口号
    private int receivePort; // 提供接收消息方的端口号
    private String ipAddress; // 提供接收方的ip地址
    private boolean status = true;
    private String name; // 给UPD对象设置一个自定义名称
    private ByteArrayOutputStream bos; // 临时缓冲

    private DatagramPacket dp;
    private  /*static*/ DatagramSocket ds;

    /**
     * @param waitPort    接收端口（写入）
     * @param receivePort 发送端口（写出/接收方）
     * @param ipAddress   接收方ip
     * @param name        给线程起个名字
     * @Description: 常用的构造器
     */
    public UDPUtils(int waitPort, int receivePort, String ipAddress, String name) {
        this.waitPort = waitPort;
        this.receivePort = receivePort;
        this.ipAddress = ipAddress;
        this.name = name;
    }

    /**
     * @param t new 一个UPDUtils对象传入，有点奇怪能用就行
     * @return
     */
    public boolean chatStart(Runnable t) {
        System.out.println("-------开启聊天-------");
        Thread receive = new Thread(t, "receive");
        receive.start();
        Thread send = new Thread(t, "send");
        send.start();
        return true;
    }

    @Override
    public void run() {
        while (status) {
            if (Thread.currentThread().getName().equals("receive"))
                try {
//                    System.out.println("等待接收消息...");
                    System.out.println(receiveMassage());

                } catch (IOException e) {
                    e.printStackTrace();
                }
            else {
//                System.out.println("等待发送消息...");
                Scanner scanner = new Scanner(System.in);
                String next = scanner.nextLine();

                /*
                待实现空字符串检测
                 */

                System.out.println("输入的内容：" + next);

                try {
                    sendMassage(next);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * @param massage 发送的消息
     * @return 成功true
     * @throws IOException
     */
    public boolean sendMassage(String massage) throws IOException {
        // 传入字符串大于64个字节
        if (massage.getBytes().length > 64 * 1024) {
            full(massage);
        } else {
            String mass = null;
            DatagramSocket ds = new DatagramSocket();
            if (name != null) {
                mass = "[" + name + "][" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss")) + "]: " + "「" + massage + "」";
            } else {
                mass = "[" + waitPort + "][" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss")) + "]: " + "「" + massage + "」";
            }
            byte[] bytes = /*massage.getBytes()*/mass.getBytes();
//        System.out.println("发送数据长度：" + bytes.length);
            dp = new DatagramPacket(bytes, bytes.length, InetAddress.getByName(ipAddress), receivePort);
//            System.out.println("包里数据的长度" + dp.getLength());
            ds.send(dp);
            ds.close();
        }

//        System.out.println("消息发送成功");
        return true;
    }


    /**
     * @return 返回从socket中得到的消息
     * @Description: 接收消息
     */
    public String receiveMassage() throws IOException {
        // 设置等待监听端口
        ds = new DatagramSocket(waitPort);
        /*
        浪费空间写法，以后优化
        */
        byte[] bytes = new byte[64 * 1024];
//        System.out.println("读入数据初始化长度：" + bytes.length);
        // 这里需要重新new包否则会出现脏数据
        DatagramPacket dp = new DatagramPacket(bytes, bytes.length);
//        System.out.println("--------接收消息");
        ds.receive(dp);
        int length = dp.getLength();
//        System.out.println("包里读入数据长度：" + length);
        ds.close();
        return new String(bytes, 0, length);
    }


    /**
     * @param massage 传入消息
     * @throws IOException
     * @Description: 当发送消息byte长度大于64*1024时调用此方法
     */
    public void full(String massage) throws IOException {
        bos = new ByteArrayOutputStream();
        byte[] bytes = massage.getBytes();
//        System.out.println("超量入读数据量：" + bytes.length);

        // 把数据写入到临时缓存中
        bos.write(bytes, 0, bytes.length);
        bos.flush();

        byte[] temp1 = bos.toByteArray(); // 超量数据临时存储
//        System.out.println("转储数据量：" + temp1.length);
        ByteArrayInputStream bis = new ByteArrayInputStream(temp1);
        byte[] temp2 = new byte[60 * 1024];
        int len = 0;

        // 把数据分成小块后逐个发送
        while ((len = bis.read(temp2)) != -1) {
//            System.out.println("循环：" + len);
            sendMassage(new String(temp2, 0, len));
        }
        bos.close();
//        System.out.println("转储成功");
    }

    //<editor-fold desc="初始化">

    /**
     * @param massage  传入需要发送的消息
     * @param ip       传入接收方的ip地址
     * @param sendPort 接收方端口号
     * @Description: UDP 发送消息
     */
    public boolean sendMassage(String massage, String ip, int sendPort) throws IOException {
        // 防止传入字符串大于64个字节
//        if (massage.getBytes().length > 64 * 1024) {
//            return false;
//        }
        String mass;
        DatagramSocket ds1 = new DatagramSocket();

        if (name != null) {
            mass = "[" + name + "][" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss")) + "]: " + "「" + massage + "」";
        } else {
            mass = "[" + waitPort + "][" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss")) + "]: " + "「" + massage + "」";
        }
        byte[] bytes = mass.getBytes();

        dp = new DatagramPacket(bytes, bytes.length, InetAddress.getByName(ip), sendPort);
        ds1.send(dp);


        ds1.close();
        return true;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public UDPUtils(int port) throws SocketException {
        this.waitPort = port;
    }

    public boolean isStatus() {
        return status;
    }

    public void setStatus(boolean status) {
        this.status = status;
    }

    public int getReceivePort() {
        return receivePort;
    }

    public void setReceivePort(int receivePort) {
        this.receivePort = receivePort;
    }

    public UDPUtils(int waitPort, int receivePort, String name, DatagramPacket dp, DatagramSocket ds) throws SocketException {
        this.waitPort = waitPort;
        this.receivePort = receivePort;
        this.name = name;
        this.dp = dp;
        this.ds = ds;
    }

    public UDPUtils(int waitPort, String name) throws SocketException {
        this.waitPort = waitPort;
        this.name = name;
    }

    public int getWaitPort() {
        return waitPort;
    }

    public void setWaitPort(int waitPort) {
        this.waitPort = waitPort;
    }

    public DatagramPacket getDp() {
        return dp;
    }

    public void setDp(DatagramPacket dp) {
        this.dp = dp;
    }

    public DatagramSocket getDs() {
        return ds;
    }

    public void setDs(DatagramSocket ds) {
        this.ds = ds;
    }

    public UDPUtils() throws SocketException {
    }


    public UDPUtils(int port, DatagramPacket dp, DatagramSocket ds) throws SocketException {
        this.waitPort = port;
        this.dp = dp;
        this.ds = ds;
    }

    public String getIpAddress() {
        return ipAddress;
    }

    public void setIpAddress(String ipAddress) {
        this.ipAddress = ipAddress;
    }
    //</editor-fold>
}

《新程序员》：云原生和全面数字化实践

50位技术专家共同创作，文字、视频、音频交互阅读