使用 ObjectStreams 通过套接字发送列表

Question

我需要在节点之间传输一个 List 并将现有的 List 替换为新节点中的新 List 来实现这一点，我正在使用 Java 中的 Sockets。

我以某种方式设法传输了数据，但只有在我终止进程时。我需要它继续运行，该过程但同时传输数据，以防任何其他新节点加入列表。 我怎样才能做到这一点？我将不得不在下载过程中介绍线程。

我让它处理文件，但现在我需要将它更改为同步列表，仅此就足够了吗？

private static List<CloudByte> cloudByteList = Collections.synchronizedList(new ArrayList<>());

这是我当前的代码：

import java.io.*;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import static FileData.getCloudByteList;
import static FileData.getFile;

public class FileData {

    private static File file;
    private static String fileName;
    private static List<CloudByte> cloudByteList = Collections.synchronizedList(new ArrayList<>());

    public FileData(String fileName) throws IOException {
        if (fileName == null) {
            this.fileName = "data2.bin";
            this.file = new File(this.fileName);
            Download.downloadFile();
        } else {
            this.file = new File(fileName);
            this.fileName = fileName;
            fillingList();
        }
    }

    public void fillingList() throws IOException {
        byte[] fileContents = Files.readAllBytes(file.toPath());
        for (int i = 0; i < fileContents.length - 1; i++) {
            cloudByteList.add(new CloudByte(fileContents[i]));
        }
    }

    public static List<CloudByte> getCloudByteList() {
        return cloudByteList;
    }

    public static File getFile() {
        return file;
    }

    public String getFileName() {
        return fileName;
    }

    public static void setFile(File file) {
        FileData.file = file;
    }

    /*--------------------------Download--------------------------*/

}

class Download extends Thread {
    static ConnectingDirectory connectingDirectory;

    @Override
    public void run() {
        try {
            downloadFile();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void downloadFile() throws IOException {

        var nodes = ConnectingDirectory.getNodes();

        Socket socket = null;

        if (getFile().exists()) {
            System.out.println("File: " + getFile() + " exists.");
            new Upload().uploadFile();
        }

        FileOutputStream fos = new FileOutputStream(FileData.getFile());
        //ObjectOutputStream oos = new ObjectOutputStream(fos);


        for (int i = 0; i < nodes.size() - 1; i++) {
            if (!(nodes.get(i).getHostPort() == ConnectingDirectory.getHostIP())) {
                System.out.println("test33123");

                ServerSocket serverSocket = new ServerSocket(nodes.get(i).getHostPort());
                System.out.println(serverSocket);
                socket = serverSocket.accept();
                System.out.println("now socket");
                System.out.println(socket);
                //socket = new Socket(nodes.get(i).getName(), nodes.get(i).getHostPort());
                //System.out.println(socket);
                int bytes = 0;
                DataInputStream ois = new DataInputStream(socket.getInputStream());
                long size = ois.readLong();
                System.out.println(size);
                byte[] buffer = new byte[100 * 10000];
                while (size > 0 && (bytes = ois.read(buffer, 0, (int) Math.min(buffer.length, size))) != -1) {
                    System.out.println("test3333");
                    fos.write(buffer, 0, bytes);
                    size -= bytes;
                }
            }
        }
    }
}

/*--------------------------Upload--------------------------*/

class Upload {

    public void uploadFile() throws IOException {
        int bytes = 0;
        var nodes = ConnectingDirectory.getNodes();

        FileInputStream fileInputStream = new FileInputStream("data.bin");
        DataInputStream ois = new DataInputStream(fileInputStream);

        if (!getFile().exists()) {
            System.out.println("File doesn't exist." + "\nDownloading the file!");
            new Download().downloadFile();
        }
        System.out.println("hello");
        for (int i = 0; i < nodes.size() - 1; i++) {
            System.out.println("hello2");
            Socket socket = new Socket(nodes.get(i).getName(), nodes.get(i).getHostPort());
            DataOutputStream dos = new DataOutputStream(socket.getOutputStream());
            dos.writeLong(new File("data.bin").length());
            byte[] buffer = new byte[100 * 10000];
            while ((bytes = ois.read(buffer)) != -1) {
                dos.write(buffer, 0, bytes);
                dos.flush();
            }
        }
    }
}

如您所见，我使用 DataInput 是因为如果我尝试使用 ObjectInputStream，则会收到损坏的标头异常。我有更多的课程要添加。正如我所说，我的目标是将“data.bin”中的数据传输到“data2.bin”文件中。我可以创建和删除它，但同时没有数据被写入/发送给它。 如何修复 CorruptedHeaderException 并让它发送内容？ 感谢所有帮助。

StorageNode 类：

import java.io.IOException;
import java.util.Scanner;
import java.util.regex.Pattern;

import static FileData.*;

public class StorageNode extends Thread {

private static int serverPort = 8080;
private static int clientPort = 8082;
private static String fileName = null;
private static String addressName = "localhost";


private static ConnectingDirectory connectingDirectory;
private static FileData fileData;
static ErrorInjection errorInjection;

public static void main(String[] args) throws IOException, InterruptedException {

   /* if (args.length > 3) {
        addressName = args[0];
        serverPort = Integer.parseInt(args[1]);
        clientPort = Integer.parseInt(args[2]);
        fileData = new FileData(args[3]);
    } else {
        fileName = null;
        fileData = new FileData(fileName);
    }*/
    connectingDirectory = new ConnectingDirectory(addressName, clientPort, serverPort);
    fileData = new FileData(fileName);
    errorInjection = new ErrorInjection();
    errorInjection.start();
    if(fileData.getFile().exists()){
        new Upload().uploadFile();
    }else {
        new Download().downloadFile();
    }
}

连接目录类

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetAddress;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;
import java.util.Scanner;

public class ConnectingDirectory {

    private String hostName;
    private static int hostIP;
    private int directoryIP;
    private InetAddress address;
    private InputStream in;
    private OutputStream out;
    private static List<Nodes> nodes = new ArrayList<>();
    List<String> nodess = new ArrayList<>();
    private Socket socket;
    private String sign = "INSC ";


    public ConnectingDirectory(String hostName, int hostIP, int directoryIP) throws IOException {
        this.hostName = hostName;
        this.hostIP = hostIP;
        this.directoryIP = directoryIP;
        this.address = InetAddress.getByName(hostName);
        this.socket = new Socket(address, directoryIP);
        signUp();
        askConnectedNodes();
    }

    public void signUp() throws IOException {
        System.out.println("You are connecting to the following address: " + hostIP + "\n");
        System.out.println("The port you are connected to: " + socket.getPort() + "\n");
        in = socket.getInputStream();
        out = socket.getOutputStream();
        out.write(generateSignUp(address, hostIP).getBytes());
        out.flush();
    }

    public String generateSignUp(InetAddress address, int hostIP) {
        String signUpString = sign + address + " " + hostIP + "\n";
        return signUpString;
    }

    public void askConnectedNodes() throws IOException {
        String directoryNodesAvailable;
        String a = "nodes\n";
        out.write(a.getBytes());
        out.flush();
        Scanner scan = new Scanner(in);

        while (true) {
            directoryNodesAvailable = scan.nextLine();
            addExistingNodes(directoryNodesAvailable);
            //System.out.println("Eco: " + directoryNodesAvailable);
            if (directoryNodesAvailable.equals("end")) {
                out.flush();
                printNodes();
                break;
            }
        }
    }

    public void addExistingNodes(String sta) throws IOException {
        if (sta.equals("end")) return;
        if (!(nodess.contains(sta))) {
            nodess.add(sta);
            nodes.add(new Nodes(nodess.get(nodess.size() - 1)));
        }
        return;
    }

    public static List<Nodes> getNodes() {
        return nodes;
    }

    public void printNodes() {
        System.out.println("Checking for available nodes: \n");
        nodes.forEach((z) -> System.out.println(z.getNode()));
    }

    public Socket getSocket() {
        return socket;
    }

    public static int getHostIP() {
        return hostIP;
    }

    public InetAddress getAddress() {
        return address;
    }
}

Answer 1

对于所有未来需要帮助的人：

发送方：

Socket socket = new Socket("localhost", hostPort);
ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream());
ByteBlockRequest bbr = new ByteBlockRequest(getStoredData());
objectOutputStream.writeObject(bbr.blocksToSend(j));

接收方：

Socket = StorageNode.getServerSocket().accept();
ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
byte[] bit = (byte[]) ois.readObject();

在我的情况下，我需要使用 byte[]，所以我必须在后面做一些额外的函数，将 Cloudbyte[] 更改为 byte[]。一旦我这样做了，我就能够使用 ObjectInput/ObjectOutput 发送数据。