基于Java套接字的文件传输应用程序客户端在调试时运行，但在正常运行时失败

Question

我接到一项任务，要创建一个类似 Dropbox 的文件多客户端单服务器文件传输应用程序。因此，我用 java 创建了一个服务器应用程序，它连续运行以接受来自客户端的连接，并为每个客户端提供自己的线程，以便它们可以同时工作。另一方面，客户端监视预先指定的文件夹中是否有任何新文件添加。一旦他们检测到新文件，就会将其发送到服务器。当服务器收到文件时，它将将该文件发送到所有连接的客户端。

我的问题是客户端代码在调试期间运行正常，但正常运行时会出现错误:错误:java.io.FileNotFoundException:D:\Distributed Systems\Labs\Lab_1\client_1\shared_directory\random.txt(该进程无法访问该文件，因为该文件正在被另一个进程使用)

经过一些调试后，我意识到发生此错误是因为客户端代码太快，这就是为什么我必须通过添加 Thread.sleep() 人为地减慢速度。

如何确保客户端在尝试将文件发送到服务器之前等待文件正确添加到受监视的文件夹中？

我正在使用 java.nio.file Watch Service API 来监视文件夹中是否有新文件或修改过的文件。

服务代码:

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.Vector;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


public class FT_Server {

    // Vector to store active clients
    static Vector<ClientHandler> connectedClients = new Vector<>();
    static ArrayList<String> clientNames = new ArrayList<>();

    public static void main(String[] args) throws Exception {
        String userName = "";

        try (ServerSocket serverSocket = new ServerSocket(59898)) {
            System.out.println("Server is running, waiting for connection");

            ExecutorService pool = Executors.newFixedThreadPool(20);

            while (true) {
                //client socket created
                Socket clientSocket = serverSocket.accept();

                //obtain input and output streams
                DataInputStream dis = new DataInputStream(clientSocket.getInputStream());
                DataOutputStream dos = new DataOutputStream(clientSocket.getOutputStream());

                userName = dis.readUTF();

                System.out.println(userName);

                // check if username already exists

                for (ClientHandler clt : FT_Server.connectedClients) {
                    clientNames.add(clt.cltName);
                }

                while (clientNames.contains(userName)) {
                    dos.writeUTF("FALSE");
                    userName = dis.readUTF();
                }

                dos.writeUTF("TRUE");

                System.out.println("Creating a new handler for this client...");

                // Create a new handler object for handling this request.
                ClientHandler clt = new ClientHandler(clientSocket, userName);

                //add the new client handler object to the vector of connected clients
                connectedClients.add(clt);
                clientNames.add(userName);

                System.out.println(connectedClients);
                System.out.println(clientNames);

                pool.execute(clt);
            }
        } catch(IOException ioe){
            ioe.printStackTrace();
        }
    }
}

class ClientHandler implements Runnable {
    DataInputStream  dis       = null;
    DataOutputStream dos       = null;
    FileOutputStream fos = null;
    Socket           cltSocket;
    String           cltName;
    FileInputStream fis = null;
    BufferedInputStream bis = null;
    InputStream in = null;
    boolean isloggedin;

    ClientHandler(Socket clientSocket, String userName) {
        this.cltSocket = clientSocket;
        this.cltName = userName;
        this.isloggedin=true;
    }

    @Override
    public void run() {
        System.out.println("inside ClientHandler class's run method");
        String fileName = "";
        int bytesRead = 0;

        while (true){
            try {
                // receive file from client
                dis = new DataInputStream(cltSocket.getInputStream());
                dos = new DataOutputStream(cltSocket.getOutputStream());

                if (dis != null)
                    System.out.println("dis not null");

                //get the name of the file
                fileName = dis.readUTF();
                System.out.println("fileName = "+fileName);

                if(fileName.equals("logout")){
                    this.isloggedin=false;
                    this.cltSocket.close();
                    break;
                }

                fos = new FileOutputStream("D:/Distributed Systems/Labs/Lab_1/server/" + fileName);

                //get the size of the file to be received
                long size = dis.readLong();
                System.out.println("size = "+size);

                byte[] buffer = new byte[(int) size];

                //write the data bytes received to a file
                while (size > 0 && (bytesRead = dis.read(buffer, 0, (int) Math.min(buffer.length, size))) != -1) {
                    fos.write(buffer, 0, bytesRead);
                    size -= bytesRead;
                }

                System.out.println("File " + fileName + " downloaded from client(" + size + " bytes read)");

            } catch (IOException e1) {
                e1.printStackTrace();
            }

            //send the file to all the connected clients
            final String FILE_TO_SEND = "D:/Distributed Systems/Labs/Lab_1/server/" + fileName;

            try {
                System.out.println("inside sending file to connected clients try section");
                File myFile = new File(FILE_TO_SEND);
                byte[] fileContents = new byte[(int) myFile.length()];

                fis = new FileInputStream(myFile);
                bis = new BufferedInputStream(fis);
                // disB = new DataInputStream(bis);

                if (bis != null){
                    System.out.println("bis not null");
                }

                //fill the data into the Byte array?
                bis.read(fileContents, 0, fileContents.length);

                // Sending file to each connected client
                for (ClientHandler clt : FT_Server.connectedClients) {
                    System.out.println("inside for loop");
                    if (clt.cltName != this.cltName && clt.isloggedin==true){
                        System.out.println("inside if");
                        //Send the file name to the client
                        clt.dos.writeUTF(myFile.getName());

                        //send the length of the file to the client
                        clt.dos.writeLong(fileContents.length);

                        System.out.println("Sending the file" + FILE_TO_SEND + "(" + fileContents.length + " bytes)");

                        //send the file contents to the client?
                        clt.dos.write(fileContents, 0, fileContents.length);
                        clt.dos.flush();

                        // // Sending file data to the client?
                        // os.write(fileContents, 0, fileContents.length);
                        // os.flush();
                        System.out.println("File sent to client = "+clt.cltName);
                    }
                }
            } catch (Exception e) {
                System.out.println("Error: " + e + "for client socket: " + cltSocket);
            }
        }
        try {
            System.out.println("inside finally");
            dis.close();
            dos.close();
            fos.close();
            fis.close();
            bis.close();
            cltSocket.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

客户端代码:

import java.io.*;
import java.net.Socket;
import java.net.UnknownHostException;
import java.nio.file.*;
import java.util.Scanner;

public class FT_Client_1 {
    Socket clientSocket = null;
    DataInputStream dis = null;
    DataOutputStream dos = null;

    public static void main(String[] args) throws Exception {

        FT_Client_1 clt = new FT_Client_1();

        clt.startConnection();

        clt.runClient();

//        clt.closeConnection();
    }

    public void runClient() {
        System.out.println("inside runClient()");
        //monitor the shared directory and send any new files to the server
        MonitorSharedDirectory mon = new MonitorSharedDirectory();
        Thread t1 = new Thread(mon);
        t1.start();

        // Receive any files sent by the server
        receiveFileFromServer rec = new receiveFileFromServer();
        Thread t2 = new Thread(rec);
        t2.start();
    }

    public void startConnection() throws UnknownHostException, IOException {
        System.out.println("inside startConnection()");
        String username = "";
        Boolean valid = true;

        // taking username as input from the user
        Scanner sc = new Scanner(System.in);

        System.out.println("Enter a username:");
        username = sc.nextLine();

        //creating client socket
        clientSocket = new Socket("127.0.0.1", 59898);

        //getting the data input and output stream using client socket
        dos = new DataOutputStream(clientSocket.getOutputStream());
        dis = new DataInputStream(clientSocket.getInputStream());

        dos.writeUTF(username);
        System.out.println("after sending username to the server");

        // Checking if server accepted the username
        do {
            String serverReply = dis.readUTF();
            if (serverReply == "FALSE"){
                // disconnect from server socket TODO
                System.out.println("Given Username is already in use, please provide another Username");
                username = sc.nextLine();
                dos.writeUTF(username);
            }else {
                valid = false;
            }
        }while (valid);

        System.out.println("after while loop to check if username is unique");
        sc.close();
        System.out.println("client " + username + " has been connected to the server");
    }

    public class MonitorSharedDirectory implements Runnable {
        FileInputStream fis = null;
        BufferedInputStream bis = null;

        @Override
        public void run() {
            System.out.println("inside MonitorSharedDirectory class's run method");

            try{
                System.out.println("inside MonitorSharedDirectory try section");

                Path watchFolder = Paths.get("D:/Distributed Systems/Labs/Lab_1/client_1/shared_directory/");
                WatchService watchService = FileSystems.getDefault().newWatchService();
                watchFolder.register(watchService, StandardWatchEventKinds.ENTRY_CREATE, StandardWatchEventKinds.ENTRY_MODIFY); //add ENTRY_MODIFY to this to monitor for file modifications

                boolean valid = true;

                do {
                    WatchKey watchKey = watchService.take();

                    for (WatchEvent event : watchKey.pollEvents()) {

                        WatchEvent.Kind kind = event.kind();
                        if (StandardWatchEventKinds.ENTRY_CREATE.equals(kind) || StandardWatchEventKinds.ENTRY_MODIFY.equals(kind)) {
                            String fileName = event.context().toString();
                            System.out.println("File Created:" + fileName);

                            int attempts = 0;
                            while(dis.available() == 0 && attempts < 1000)
                            {
                                attempts++;
                                Thread.sleep(5);
                            }

                            // sending new file to server
                            File myFile = new File("D:/Distributed Systems/Labs/Lab_1/client_1/shared_directory/" + fileName);
                            byte[] fileContents = new byte[(int) myFile.length()];

                            fis = new FileInputStream(myFile);
                            bis = new BufferedInputStream(fis);
                            DataInputStream dis = new DataInputStream(bis);

                            dis.readFully(fileContents, 0, fileContents.length);

                            dos.writeUTF(myFile.getName());
                            dos.writeLong(fileContents.length);
                            dos.write(fileContents, 0, fileContents.length);
                            dos.flush();
                            System.out.println("File "+fileName+" sent to Server.");


//                            //fill the data into the Byte array?
//                            bis.read(fileContents, 0, fileContents.length);
//
//                            dos.writeUTF(myFile.getName());
//
//                            //send the length of the file to the client
//                            dos.writeLong(fileContents.length);
//
//                            System.out.println("Sending the file " + myFile + " (" + fileContents.length + " bytes)");
//
//                            //send the file contents to the server?
//                            dos.write(fileContents, 0, fileContents.length);
//                            dos.flush();

                            if (fis != null)
                                fis.close();
                            if (bis != null)
                                bis.close();
                            if (dis != null)
                                dis.close();
                        }
                    }
                    valid = watchKey.reset();
                } while (valid);
            }catch(Exception e){
                System.out.println("Error Prem: " + e );
            }finally {
                //if (dos != null)
                //dos.close();
                try {
                    if (fis != null)
                        fis.close();
                    if (bis != null)
                        bis.close();
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }
    }

    public class receiveFileFromServer implements Runnable {
        FileOutputStream fos = null;

        @Override
        public void run() {
            System.out.println("inside receiveFileFromServer class's run method");

            while (true){
                try {
                    System.out.println("inside receiveFileFromServer try section");

                    // receive file from server
                    String fileName = dis.readUTF();

                    System.out.println(fileName);

                    fos = new FileOutputStream("D:/Distributed Systems/Labs/Lab_1/client_1/shared_directory/" + fileName);

                    int bytesRead = 0;

                    long size = dis.readLong();

                    byte[] buffer = new byte[(int) size];

                    while (size > 0 && (bytesRead = dis.read(buffer, 0, (int) Math.min(buffer.length, size))) != -1) {
                        fos.write(buffer, 0, bytesRead);
                        size -= bytesRead;
                    }
                    System.out.println("File " + fileName + " downloaded from server(" + size + " bytes read)");

                    if (fos != null)
                        fos.close();

                } catch (Exception e) {
                    System.out.println("Error: " + e);
                    e.printStackTrace();
                }
            }
        }
    }

    public void closeConnection() throws IOException {
        System.out.println("inside closeConnection()");
        if (dis != null)
            dis.close();
        if (dos != null)
            dos.close();
        if (clientSocket != null)
            clientSocket.close();
    }
}

Answer 1

简单的方法 - 您可以通过在服务器上保存小 dB 的文件寄存器来处理这个问题，其中包含文件标记和文件大小，我们称之为元数据。

这样，万一文件将来发生更改并第二次发送更改后的内容，您可以查找元数据。

如果将相同的文件发送到服务器，服务器会拒绝该文件，并且不会将其传播到其他客户端。

就我而言，我使用的是 aws sqs，并且我查找消息 ID 以检查重复文件。

我希望这种方法有所帮助。