java - 在 Java 服务器中的客户端线程之间切换

Question

我正在开展一个项目，试图让几个人能够控制机器人 ARM 。为此，他们必须连接到 Java 服务器，然后该服务器将命令发送到机器人屏幕以进行视频 session 。 我试图为每个客户端建立一个线程，然后我希望能够根据声音在不同客户端之间切换，因为我希望扬声器能够控制机器人。 客户端都提供位置数据和 kinect 获取的声音级别，并以字符串的形式发送到服务器。 我在执行切换时遇到问题。目前，它们似乎在来回切换，这使得机器人变得失控。

有没有一种好方法可以相互比较线程，找到合适的线程，切换到该线程，同时检查其他线程以查看它们是否或何时成为最合适的线程？同时还要检查其他客户端是否尝试连接到服务器？

感谢您的帮助。

我还包含了我的代码，以防您想查看它并获得更好的想法。

这是服务器类:

import java.io.IOException;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.util.ArrayList;
import java.util.Hashtable;

public class MultiThreadedServer implements Runnable {
    protected int       serverPort  = 8888;
    protected ServerSocket  serverSocket    = null;
    protected boolean   isStopped       = false;
    protected Thread        runningThread   = null;
    protected Thread        clientThread    = null;
    protected Thread        threadThread    = null;
    private Hashtable<Long, WorkerRunnable> Users = new Hashtable<Long, WorkerRunnable>();
    private ArrayList<Thread> ClientThreads = new ArrayList<Thread>();
    private WorkerRunnable  client          = null;
    private ThreadHandler   threadHandler   = null;
    private int             sound_max       = 0;
    private boolean         once            = true;

    public MultiThreadedServer (int port) {
         this.serverPort = port;
    }

    public void run() {
    synchronized(this) {
        this.runningThread = Thread.currentThread();
    }
    openServerSocket();
    threadHandler = new ThreadHandler();
    while( !isStopped() ) {
        Socket clientSocket = null;
        try {
            System.out.println(InetAddress.getLocalHost());
            clientSocket = this.serverSocket.accept();          // Connect to clients
        } catch (SocketTimeoutException e) {

        } catch (IOException e) {
            if( isStopped() ) {
                System.out.println("Server Stopped");
                return;
            }
            throw new RuntimeException("Error accepting client connection", e);
        }

        client = new WorkerRunnable(clientSocket, "Multithreaded Server");//Class does client work
        clientThread = new Thread(client);          // Make a thread for each client
        clientThread.start();                       // start thread

        threadHandler.setUp(client, clientThread);  // Set up the thread handler
        if ( once == true) {                        // make sure the threadHandler thread is only created once
            threadThread = new Thread(threadHandler);
            threadThread.start();
            once = false;
        }   
    }
    System.out.println("Server Stopped");
}

/**
 * Check if the socket is stopped
 * @return true if the socket is stopped
 */
private synchronized boolean isStopped() {
    return this.isStopped;
}

/**
 * Stop and close the socket
 */
public synchronized void stop() {
    this.isStopped = true;
    try {
        this.serverSocket.close();
    } catch (IOException e) {
        throw new RuntimeException("Error closing server", e);
    }
}

    /**
     * Open server socket
     */
    private void openServerSocket() {
        try {
            this.serverSocket = new ServerSocket(this.serverPort);
        } catch (IOException e) {
            throw new RuntimeException("Cannot open port 8888", e);
        }
    }
}

这是 Worker 类，处理来自客户端的数据:

import gnu.io.NoSuchPortException;
import java.io.BufferedReader;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.net.Socket;

public class WorkerRunnable implements Runnable {

    protected Socket clientSocket   = null;
    protected String serverText     = null;
    private BufferedReader inFromClient;
    private DataOutputStream outToClient;
    private int[] currentPos = new int[6];
    private boolean connected = false;
    static TwoWaySerialComm serialCom = null;
    static MultiServoState mState;
    static int sound_average;
    int[] degrees = new int[7];
    int count = 0;

    public WorkerRunnable(Socket clientSocket, String serverText) {
        this.clientSocket = clientSocket;
        this.serverText = serverText;
        initCurrentPos();
        if (serialCom == null) {
             serialCom = new TwoWaySerialComm();
        }
        try {
            if (!serialCom.isConnected("COM5")) {
                try {
                    serialCom.connect("COM5");
                } catch (Exception e) {
                // TODO Auto-generated catch block
                    e.printStackTrace();
                }
                mState = new MultiServoState(serialCom);
            }
        } catch (NoSuchPortException e) {
        // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    public void run() {
        try {
            work();
        } catch (InterruptedException e) {
        // TODO Auto-generated catch block
        }
    }

    public void work() throws InterruptedException {
        try {
            InputStream input = clientSocket.getInputStream();
            OutputStream output = clientSocket.getOutputStream();
            inFromClient = new BufferedReader(new InputStreamReader(input));
            outToClient = new DataOutputStream(output);
            long time = System.currentTimeMillis();
            updateData();
            String message = null;
            long endTime = System.currentTimeMillis() + 2000;
            while ((message = (String) inFromClient.readLine()) != null) {

                System.out.println("Message Received: " + message);
                parse(message);

                sound_average = degrees[6];
                //
                // Send the positional data to the robot
                //
                mState.runServo(degrees[0], degrees[1], degrees[2],
                    degrees[3], degrees[4], degrees[5]);


                //
                // Send a response information to the client application
                //
                currentPos[0] = mState.getCurrentPos(0);
                currentPos[1] = mState.getCurrentPos(1);
                currentPos[2] = mState.getCurrentPos(2);
                currentPos[3] = mState.getCurrentPos(3);
                currentPos[4] = mState.getCurrentPos(4);
                try {
                    updateData();
                } catch (IOException e) {
                // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
            System.out.println("Request processed: " + time);
        } catch (IOException e) {
            // report exception somewhere
            e.printStackTrace();
        } 
    }

          /**
          * Initiate the robot's starting position.
          */
          public void initCurrentPos()
          {
              currentPos[0] = 100;
    currentPos[1] = 100;
    currentPos[2] = 100;
    currentPos[3] = 100;
    currentPos[4] = 100;
    currentPos[5] = 0;
          }

    /**
    * Send the data to the client
    * 
    * @throws IOException
    */
    public void updateData() throws IOException {
    String sentence = Integer.toString(currentPos[0]) + ", " + 
                      Integer.toString(currentPos[1]) + ", " +
                      Integer.toString(currentPos[2]) + ", " + 
                      Integer.toString(currentPos[3]) + ", " + 
                      Integer.toString(currentPos[4]) + "." + "\n";
        outToClient.flush();
        outToClient.writeBytes(sentence);
    }

    /**
     * Get the clients sound average
     * @param message
     */
    public int getSoundAverage() {
        return sound_average;
    }

    public void parse(String message) {
        if (message != null) {
            char c;
            StringBuilder sb = new StringBuilder(4);
            int j = 0;
            boolean help = false;

            for (int i = 0; i < message.length(); i++) {
                c = message.charAt(i);
                if (Character.isDigit(c)) {
                    sb.append(c);
                    help = true;
                }
                if (!Character.isDigit(c) && help == true) {
                    degrees[j] = Integer.parseInt(sb.toString());
                    j++;
                    help = false;
                    sb.delete(0, sb.length());
                }
            }
        }
        System.out.println("Waiting for client message...");
    }

    /**
     * Close all connections
     */
    public void close() {
        if (connected) {
            synchronized (this) {
                connected = false;
            }
            if (outToClient != null) {
                try {
                    outToClient.close();
                    synchronized (this) {
                        outToClient = null;
                    }
                } catch (IOException e) {
                    // there is nothing we can do: ignore it
                }
            }

            if (inFromClient != null) {
                try {
                    inFromClient.close();
                    synchronized (this) {
                        inFromClient = null;
                    }
                } catch (IOException e) {
                    // there is nothing we can do: ignore it
                }
            }

            if (clientSocket != null) {
                try {
                    clientSocket.close();
                    synchronized (this) {
                        clientSocket = null;
                    }
                } catch (IOException e) {
                    // there is nothing we can do: ignore it
                }
            }
        }
    }

    public void returnThread() {
        return;
    }
}

最后一个类是线程处理程序，我在其中尝试比较声音级别并生成除最响亮的线程之外的所有线程:

import java.net.ServerSocket;
import java.util.ArrayList;
import java.util.Hashtable;

import com.research.aserver.WorkerRunnable;


public class ThreadHandler implements Runnable {

    protected boolean   isStopped       = false;
    protected Thread        runningThread   = null;
    protected Thread        clientThread    = null;
    private Hashtable<Long, WorkerRunnable> Users = new Hashtable<Long, WorkerRunnable>();
    private ArrayList<Thread> ClientThreads = new ArrayList<Thread>();
    private WorkerRunnable  client      = null;
    private int     sound_max       = 0;
    private int     index       = 0;

    public ThreadHandler() {
    }

    public void setUp(WorkerRunnable client, Thread clientThread) {
        this.client = client;
        this.clientThread = clientThread;
        Users.put(clientThread.getId(), this.client);   // Place clients in a list with its thread ID as key
        ClientThreads.add(this.clientThread);           // List of client threads
    }

    @Override
    public void run() {
        long endTime = System.currentTimeMillis() + 2000;       // Help variable to check every 2 sec
        while (!Users.isEmpty() && !ClientThreads.isEmpty()) {

            for (int i = 0; i < ClientThreads.size(); i++) {    // Remove clients and threads if no longer active
                if (!ClientThreads.get(i).isAlive()) {
                    Users.remove(ClientThreads.get(i).getId());
                    ClientThreads.get(i).interrupt();
                    ClientThreads.remove(i);
                }
            }
            if(System.currentTimeMillis() >= endTime) { // Do work every 2 sec
                for (int i = 0; i < ClientThreads.size(); i++) {    // Get the client with the loudest sound
                    if (sound_max < Users.get(ClientThreads.get(i).getId()).getSoundAverage()) {
                        sound_max = Users.get(ClientThreads.get(i).getId()).getSoundAverage();
                        index = i;
                    }
                }

                for (int i = 0; i < ClientThreads.size(); i++) {    // yield all threads that are not the loudest
                    if (Users.get(ClientThreads.get(index).getId()) != Users.get(ClientThreads.get(i).getId())){
                        ClientThreads.get(i).yield();
                        index = 0;
                    }
                }
                endTime = System.currentTimeMillis() + 2000;                        // update time
            }
            sound_max = 0;
        }
    }
}

Answer 1

一个想法可能是使用 PriorityBlockingQueue并为每个输入定义一个质量值，然后在列表中自动按质量排序。

使用它，您的消费者线程可以简单地获取队列中的第一个并处理它，知道它是最合适的，而生成器线程可以简单地将所有输入扔到队列中。