Сам текущий поток ждет бесконечно, пытаясь вызвать wait () в другом потоке - PullRequest
0 голосов
/ 01 ноября 2018

Я пытаюсь заставить другой поток ждать в приведенном ниже коде, но сам мой текущий поток ждет бесконечно. Ниже приведены два java-класса Server.java, которые порождают работающие экземпляры ServerService.java. Когда такой запущенный экземпляр «ServerService.java» вызывает метод постановки в очередь Server.java.Server.java, это заставит такой поток вызываемых ждать. Но, похоже, мой поток Server.java сам ждет бесконечно

Server.java

import java.io.DataInputStream;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Server{
    private ServerSocket server=null;
    public static Map<Socket,String> clientsConnected=null;
    public static Map<Socket,Runnable> clientsAndThreads=null;
    public static ExecutorService executor=null;
    public static List<Runnable> requestQueue=null;
    public static Map<Runnable,Integer> threadAndRespectiveTime=null;
    /*
     * Contructor
     */
    Server(){
        clientsConnected=new HashMap<Socket,String>();
        clientsAndThreads=new HashMap<Socket,Runnable>();
        threadAndRespectiveTime=new HashMap<>();
        requestQueue=new ArrayList<>();
    }
    /*
     * Accepts connections from clients continually till the server is UP(max 10 clients)
     */
    public void acceptConnection(){
        try{
            executor=Executors.newFixedThreadPool(10);
            new Thread(new Runnable() {
                @Override
                public void run() {
                    Socket client=null;
                    while(server.isBound()){
                        try{
                            client=server.accept();
                            DataInputStream di= new DataInputStream(client.getInputStream());
                            String msg=di.readUTF();
                            clientsConnected.put(client, getMessage(msg));
                            ServerWindow.write(msg);
                            Runnable service= new ServerService(client,getMessage(msg));
                            executor.execute(service);
                            clientsAndThreads.put(client, service);
                        }catch(Exception e){
                            System.err.println("error occurred while accepting connections");
                        }
                    }
                }
            }).start();
            System.out.println("now dequeuing");
            while(true){
                dequeue();
            }
        }catch(Exception e){
            System.err.println("Server:error while accepting connections"+e.getMessage());
        }
    }
    public static void enqueue(Socket clientSocket,Integer secondsToWait){
        try{
            Runnable respectiveThread = clientsAndThreads.get(clientSocket);
            threadAndRespectiveTime.put(respectiveThread, secondsToWait);
            System.out.println("making thread wait");
            synchronized (respectiveThread) {
                respectiveThread.wait();
            }
            requestQueue.add(respectiveThread);
            System.out.println("done enqueuing");
        }catch(Exception e){
            e.printStackTrace();
        }
    }
    public static void dequeue() throws InterruptedException{
        while(!requestQueue.isEmpty()){
            Runnable currentThread=requestQueue.get(0);
            Integer timeToWait=threadAndRespectiveTime.get(currentThread);
            Thread.sleep(timeToWait * 1000);
            requestQueue.remove(0);
            System.out.println("wait is complete now notifying thread");
            synchronized (currentThread) {
                currentThread.notify();
            }
        }
    }
    /*
     * This method takes out actual message from http format
     */
    public String getMessage(String str){
        return str.substring(str.indexOf("message:")+8, str.length());
    }
    /*
     * Starts the server listening to port 4000
     */
    public void start_server(){
        try{
            if(server==null || !server.isBound()){
                server = new ServerSocket(4000);
            }
            acceptConnection();
        }catch(Exception e){
            System.err.println("Server:error occurred while server start"+e.getMessage());
        }
    }
    /*
     * Closes client sockets of every connected client, shuts down the thread executor that serves clients
     */
    public void stop_server() throws IOException{
        Iterator it=clientsConnected.entrySet().iterator();
        while(it.hasNext()){
            Map.Entry e= (Map.Entry)it.next();
            Socket toBeClosed=(Socket)e.getKey();
            toBeClosed.close();
        }
        executor.shutdownNow();
        server.close();
    }
}

Ниже приведен класс, который порождается сервером.java

ServerService.java

`import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.Socket;
import java.util.Date;
/*
 * This class serves the client
 */
public class ServerService extends Server implements Runnable{
    private Socket client=null;
    private String clientBeingServed=null;
    private DataOutputStream dout=null;
    private DataInputStream din=null;
    /*
     * This is construcor that takes client sockte that already has been connected to server and client name.
     * It initializes and input and output streams for serving the respective client
     */
    public ServerService(Socket client,String name) throws IOException {
        this.client=client;
        this.clientBeingServed=name;
        dout=new DataOutputStream(client.getOutputStream());
        din=new DataInputStream(client.getInputStream());
    }
    /*
     * takes out actual message sent by client from its http format
     */
    public String getMessage(String str){
        //System.out.println("returning\n"+str.substring(str.indexOf("message:")+8, str.length()));
        return str.substring(str.indexOf("message:")+8, str.length());
    }
    /*
     * This method converts a message string into HTTP formatted string
     */
    public String getHttpMessage(String msg){
        String str="POST Http/1.1 \n" + "Host: www.uta.com \n" + "User-Agent: Mozilla/5.0 \n"
                + "Content=type: application/x-www-form-urlencoded \n" + "Content-Length: " + msg.length() + " \n"
                + "Date:" + new Date() + " \n" + "message:" + msg;
        return str;
    }
    /*
     * This method execute when thread for this class is executed from Server.java file after connection is accepted
     */
    @Override
    public void run() {
        int waitTime=0;
        try{
            while(client.isConnected()){
                    String msg=din.readUTF();
                    ServerWindow.write(msg);
                    waitTime=Integer.parseInt(getMessage(msg));
                    System.out.println("Equeing:"+clientBeingServed);
                    Server.enqueue(client, waitTime);
                    ServerWindow.write("Served client:"+clientBeingServed);
                    dout.writeUTF(getHttpMessage("Server waited "+waitTime+" seconds for "+clientBeingServed));
                    dout.flush();
            }
            client.close();
        }catch(Exception e){
            System.err.println("ServerService:error serving client"+clientBeingServed+e.getMessage());
        }
    }
}`

1 Ответ

0 голосов
/ 02 ноября 2018

Вместо вызова wait для порожденного потока, экземпляр потока My ServerService ждет себя после вызова метода постановки в очередь Server.java. Затем позднее Server.java вызывает уведомление для возобновления потока ServerService.

Server.java

import java.io.DataInputStream;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.SynchronousQueue;

public class Server{
    private ServerSocket server=null;
    public static Map<Socket,String> clientsConnected=null;
    public static Map<Socket,Runnable> clientsAndThreads=null;
    public static ExecutorService executor=null;
    public static Queue<Thread> requestQueue=null;
    public static Map<Thread,Integer> threadAndRespectiveTime=null;
    /*
     * Contructor
     */
    Server(){
        clientsConnected=new HashMap<Socket,String>();
        clientsAndThreads=new HashMap<Socket,Runnable>();
        threadAndRespectiveTime=new HashMap<>();
        requestQueue=new LinkedList<>();
    }
    /*
     * Accepts connections from clients continually till the server is UP(max 10 clients)
     */
    public void acceptConnection(){
        try{
            executor=Executors.newFixedThreadPool(10);
            new Thread(new Runnable() {
                @Override
                public void run() {
                    Socket client=null;
                    while(server.isBound()){
                        try{
                            client=server.accept();
                            DataInputStream di= new DataInputStream(client.getInputStream());
                            String msg=di.readUTF();
                            clientsConnected.put(client, getMessage(msg));
                            ServerWindow.write(msg);
                            Runnable service= new ServerService(client,getMessage(msg));
                            executor.execute(service);
                        }catch(Exception e){
                            System.err.println("error occurred while accepting connections");
                        }
                    }
                }
            }).start();
            new Thread(new Runnable() {

                @Override
                public void run() {
                    try {
                        while(true){
                            Server.dequeue();
                        }
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
            }).start();
        }catch(Exception e){
            System.err.println("Server:error while accepting connections"+e.getMessage());
        }
    }
    public static synchronized void enqueue(Thread t,Integer secondsToWait){
        try{
            System.out.println(requestQueue );
            threadAndRespectiveTime.put(t, secondsToWait);
            requestQueue.add(t);
        }catch(Exception e){
            e.printStackTrace();
        }
    }
    public static synchronized void dequeue() throws InterruptedException{
        while(!requestQueue.isEmpty()){
            Thread currentThread=requestQueue.remove();
            Integer timeToWait=threadAndRespectiveTime.get(currentThread);
            System.out.println("time to wait is:"+timeToWait);
            Thread.currentThread().sleep(timeToWait * 1000);
            synchronized (currentThread) {
                currentThread.notify();
            }
        }
    }
    /*
     * This method takes out actual message from http format
     */
    public String getMessage(String str){
        return str.substring(str.indexOf("message:")+8, str.length());
    }
    /*
     * Starts the server listening to port 4000
     */
    public void start_server(){
        try{
            if(server==null || !server.isBound()){
                server = new ServerSocket(4000);
            }
            acceptConnection();
        }catch(Exception e){
            System.err.println("Server:error occurred while server start"+e.getMessage());
        }
    }
    /*
     * Closes client sockets of every connected client, shuts down the thread executor that serves clients
     */
    public void stop_server() throws IOException{
        Iterator it=clientsConnected.entrySet().iterator();
        while(it.hasNext()){
            Map.Entry e= (Map.Entry)it.next();
            Socket toBeClosed=(Socket)e.getKey();
            toBeClosed.close();
        }
        executor.shutdownNow();
        server.close();
    }
}

ServerService.java

==================

    import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.Socket;
import java.util.Date;
/*
 * This class serves the client
 */
public class ServerService extends Server implements Runnable{
    private Socket client=null;
    private String clientBeingServed=null;
    private DataOutputStream dout=null;
    private DataInputStream din=null;
    /*
     * This is construcor that takes client sockte that already has been connected to server and client name.
     * It initializes and input and output streams for serving the respective client
     */
    public ServerService(Socket client,String name) throws IOException {
        this.client=client;
        this.clientBeingServed=name;
        dout=new DataOutputStream(client.getOutputStream());
        din=new DataInputStream(client.getInputStream());
    }
    /*
     * takes out actual message sent by client from its http format
     */
    public String getMessage(String str){
        //System.out.println("returning\n"+str.substring(str.indexOf("message:")+8, str.length()));
        return str.substring(str.indexOf("message:")+8, str.length());
    }
    /*
     * This method converts a message string into HTTP formatted string
     */
    public String getHttpMessage(String msg){
        String str="POST Http/1.1 \n" + "Host: www.uta.com \n" + "User-Agent: Mozilla/5.0 \n"
                + "Content=type: application/x-www-form-urlencoded \n" + "Content-Length: " + msg.length() + " \n"
                + "Date:" + new Date() + " \n" + "message:" + msg;
        return str;
    }
    /*
     * This method execute when thread for this class is executed from Server.java file after connection is accepted
     */
    @Override
    public void run() {
        int waitTime=0;
        try{
            while(client.isConnected()){
                    String msg=din.readUTF();
                    ServerWindow.write(msg);
                    waitTime=Integer.parseInt(getMessage(msg));
                    System.out.println("Equeing:"+clientBeingServed);
                    Server.enqueue(Thread.currentThread(), waitTime);
                    System.out.println("before going to sleep");
                    synchronized (Thread.currentThread()) {
                        Thread.currentThread().wait();
                    }
                    System.out.println("after sleeping");
                    ServerWindow.write("Served client:"+clientBeingServed);
                    dout.writeUTF(getHttpMessage("Server waited "+waitTime+" seconds for "+clientBeingServed));
                    dout.flush();
            }
            client.close();
        }catch(Exception e){
            System.err.println("ServerService:error serving client"+clientBeingServed+e.getMessage());
        }
    }
}
...