Java多线程服务器-客户端通信

Question

我遇到了一个令人沮丧的问题，我正在尝试为大学考试做准备 - 应该有一堆客户端通过单独服务器上的 tcp 套接字访问资源，这会同步使用权。我有一个基本的入口点类，主要是实例化客户端或服务器，以及一个用于被序列化并通过套接字传递的对象的类：

DataPacket.class 来源：

import java.io.*;
import java.util.Date;

public class DataPacket implements Serializable{
    /**
     * 
     */
    private static final long serialVersionUID = 1L;
    final public String payload;
    final public String creator;
    final public Date timestamp;

    public DataPacket(String creator, String payload){
        this.creator = creator;
        this.payload = payload;
        this.timestamp = new Date();
    }

}

client.class 源在这里：

import java.io.*;
import java.net.*;
import java.util.Random;

public class client extends Thread implements Serializable{
    /**
     * 
     */
    private static final long serialVersionUID = 1L;
    private String name;
    public client(String name){
        this.name = name;
    }
    public void run(){
        try {
            Socket sock = new Socket(InetAddress.getLocalHost(), 12345);
            ObjectInputStream in = new ObjectInputStream(sock.getInputStream());
            ObjectOutputStream out = new ObjectOutputStream(sock.getOutputStream());
            Random rand = new Random();

// obviously the only reason to have this loop here is to initiate a bunch of access 
// requests to the server, but for debugging purposes I've removed the extra cycles
// and stuck to only one;

            for(int i=0; i<1; i++){ 
                DataPacket data = new DataPacket(this.name, Integer.toString(rand.nextInt()));
                out.writeObject(data);
                out.flush();
            }
            sock.close();
        } catch (UnknownHostException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }

    }
}

最后 server.class 源代码在这里：

import java.io.*;
import java.net.*;

public class server extends Thread{
    private int port;
    private DataPacket dp;
    public boolean running=false;

    public void updateData(DataPacket dp){
        this.dp=dp;
    }

    public server(int port){
        this.port = port;
        }

    public void run(){
        this.running = true;
        while (running){
            openSocket();
            }
        }

    public void openSocket(){
        try {
            ServerSocket s = new ServerSocket(this.port, 50, InetAddress.getLocalHost());
            while (true){
                Socket incomming = s.accept();
                ObjectInputStream in = new ObjectInputStream(incomming.getInputStream());
                ObjectOutputStream out = new ObjectOutputStream(incomming.getOutputStream());
                synchronized (dp){
                    while ((dp = (DataPacket) in.readObject()) != null) {
                        System.out.println(dp.creator + "passed a frame @" + dp.timestamp + " :"+dp.payload);
                    } while (in.readObject() == null){
                        Thread.sleep(100);
                        System.out.print(".");
                    }
                }
                s.close();
            }           
        } catch (Exception e){
            e.printStackTrace();
        }
    }
}

现在我不知道为什么，但显然客户端也需要定义一个输入流（不做任何从服务器到客户端的对话，但除非我声明，否则它会因一堆拒绝连接异常而爆炸一.为了以防万一，我为了安全起见并在服务器上声明了一个输出。不幸的是，尽管客户端似乎没有通过套接字传递对象，而服务器没有得到它，或者我失败了序列化和反序列化有些悲惨..

虽然我在做这件事，但我希望能得到一些关于服务器上数据访问同步的建议 - 这个“同步 (dp)”会完成这项工作，还是我需要将线程放在客户端手动进入等待队列，并使用 notify() 手动让其中一个客户端传输其序列化数据？

提前致谢！ vlex

Answer 1

嗯，我花了几天的额外时间来处理它，并重写了几次，现在可以工作了。这是我的启用多线程服务器的示例：

serverentry.class：

package server2;

public class serverentry {

    public static void main(String[] args) {
        server[] serv = new server[10];
        for (int i=0; i<10; i++){
            serv[i] = new server(12345+i);
            serv[i].start();
        }
    }
}

server.class：

package server2;

import java.io.*;
import java.net.*;
import java.util.*;

public class server extends Thread implements Serializable{
    /**
     * 
     */
    private static final long serialVersionUID = 1L;
    private int port;
    private static String data = "nothing yet";
    public server(int port){
        this.port = port;
        System.out.println("Server thread["+(this.port-12345)+"] started");
    }

    public void run(){
        try{
            ServerSocket sock = new ServerSocket(this.port, 50, InetAddress.getLocalHost());
            Socket accepted = sock.accept();                
            BufferedReader br = new BufferedReader(new InputStreamReader(accepted.getInputStream()));
            while(br.readLine() !=null){
                synchronized(data){
                    updateData(br.readLine());
                }
                if (br.readLine() == null) break;
            }
            sock.close();
        } catch (Exception e) {
            e.printStackTrace();
        }       
    }

    private void updateData(String streamObj){
        String[] streamArr = streamObj.split("&");
        data = streamArr[1];
        System.out.println("Data updated to "+data+" by "+streamArr[0]+" @ "+ new Date());
    }
}

cliententry.class：

package client2;

public class cliententry {

    public static void main(String[] args) {
        client2[] cli = new client2[10];
        for(int i=0; i<10; i++){
            cli[i] = new client2(i);
            cli[i].start();
        }
    }
}

client2.class：

package client2;

import java.io.*;
import java.net.*;
import java.util.*;

public class client2 extends Thread implements Serializable{
    /**
     * 
     */
    private static final long serialVersionUID = 1L;
    private String payload;
    private int number; 
    private Random rand;
    public client2(int num){
        rand = new Random();
        this.number = num;
    }

    public void run(){
        try{
            Socket sock = new Socket(InetAddress.getLocalHost(),(12345+this.number));
            PrintWriter pw = new PrintWriter(new OutputStreamWriter(sock.getOutputStream()));
            for (int i=0; i<30; i++){
                String random = Integer.toString(this.rand.nextInt());
                this.payload = "thread[" + this.number + "]&"+random; 
                pw.write(this.payload);
                pw.write("\n");
                pw.flush();
                }
            sock.close();
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

还有一次我重写了它，我使用了 run() 方法而不是 start()，这后来解释了为什么我在 10 次更新中只有 3 次左右 - 运行在同一个线程中启动逻辑，而start 实例化另一个线程。现在这一切都像瑞士时钟一样工作:)

干杯， vlex