Android 两个线程在同一个套接字上读/写

Question

我正在尝试使用托管服务器的服务，每当它从一个客户端接收数据时，它就会将数据发送到另一台服务器。两者都通过无限期保持打开的 tcp 套接字连接。我在实现可以正确读写的单个 tcp 套接字时遇到了问题。

我从两端接收 XML，并且它们定义明确。对收到的 xml 进行了一些处理，需要将其添加到处理其订单的队列中。

理想情况下，任一方向的连接都应无限期保持打开状态。

但到目前为止，我看到套接字只是不断关闭此服务，而 ServerCode 正在关闭套接字，我不知道为什么。

有没有办法建立到我的两个端点的连接并无限期地保持套接字打开？

public class routing extends Service {

    private static final String TAG = "[RoutingService]";

    private final IBinder mBinder = new RoutingBinder();
    private final ScheduledThreadPoolExecutor mRoutingThreadPool = new ScheduledThreadPoolExecutor(2);

    private boolean running = false;
    private URI serverAddress;
    private URI clientAddress;

    private Thread serverServiceThread = new ClientService();
    private Thread clientServiceThread = new ServerService();

    private PriorityBlockingQueue<String> clientQueue;
    private PriorityBlockingQueue<String> serverQueue;

    public void setClientAddress(URI testServer) {
        this.serverAddress = testServer;
        this.mRoutingThreadPool.remove(clientServiceThread);
        this.mRoutingThreadPool.scheduleWithFixedDelay(clientServiceThread, 0, 100, TimeUnit.MILLISECONDS);
    }

    public URI getServerAddress() {
        return serverAddress;
    }

    public void setServerAddress(URI testServer) {
        startRunning();
        this.serverAddress = testServer;
        this.mRoutingThreadPool.remove(serverServiceThread);
        this.mRoutingThreadPool.scheduleWithFixedDelay(serverServiceThread, 0, 100, TimeUnit.MILLISECONDS);
    }

    public void startRunning() {
        running = true;
    }

    public void stopRunning() {
        running = false;
    }

    @Override
    public void onCreate() {
        super.onCreate();
        serverQueue = new PriorityBlockingQueue<>();
        clientQueue = new PriorityBlockingQueue<>();
    }

    @Override
    public void onDestroy() {
        stopRunning();
        super.onDestroy();
    }

    @Nullable
    @Override
    public IBinder onBind(Intent intent) {
        return mBinder;
    }

    @Override
    public int onStartCommand(@Nullable Intent intent, int flags, int startId) {
        clientAddress = URI.create("127.0.0.1:8054");
        serverAddress = URI.create("192.168.2.1:7087");

        startRunning();
        setClientAddress(clientAddress);
        setServerAddress(serverAddress);

        return Service.START_STICKY;
    }

    public class RoutingBinder extends Binder {
        public routing getService() {
            return routing.this;
        }
    }

    class ClientService extends Thread {
        private Socket socket;

        private Runnable ClientReader = new Runnable() {
            @Override
            public void run() {
                if (socket != null && socket.isConnected()) {
                    try (InputStreamReader sr = new InputStreamReader(socket.getInputStream())) {
                        StringBuilder xml = new StringBuilder();
                        char[] buffer = new char[8192];
                        String content = "";
                        int read;
                        while ((read = sr.read(buffer, 0, buffer.length)) != -1) {
                            serverQueue.add(new String(buffer));
                        }
                    } catch (IOException e) {
                        Log.e("clientReader", "Error in testReading Thread.", e);
                    }
                }
            }
        };

        private Runnable ClientWriter = new Runnable() {
            @Override
            public void run() {
                if (socket != null && socket.isConnected()) {
                    while (serverQueue != null && !serverQueue.isEmpty()) {
                        try (OutputStream os = socket.getOutputStream()) {
                            String xml = serverQueue.poll();
                            os.write(xml.getBytes());
                            os.flush();
                        } catch (IOException e) {
                            Log.e("clientWriter", "Error in testReading Thread.", e);
                        }
                    }
                }
            }
        };

        @Override
        public void run() {
            try (ServerSocket server = new ServerSocket(clientAddress.getPort())) {
                try (Socket socket = server.accept()) {
                    socket.setSoTimeout(0);
                    Log.d("SOCKET", String.format("Local Port: %s. Remote Port: %s", socket.getLocalPort(), socket.getPort()));
                    this.socket = socket;
                    //Make the Threads
                    Thread reader = new Thread(ClientReader);
                    Thread writer = new Thread(ClientWriter);
                    //Start the Threads
                    reader.start();
                    writer.start();
                    //Start the Server
                    startRunning();
                    //Join on the Threads so this driver thread will wait until they finish.
                    reader.join();
                    writer.join();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            stopRunning();
        }
    }

    class ServerService extends Thread {
        private Socket socket;

        private Runnable ServerReader = new Runnable() {
            @Override
            public void run() {
                if (socket != null && !socket.isClosed()) {
                    try (InputStreamReader sr = new InputStreamReader(socket.getInputStream())) {
                        StringBuilder xml = new StringBuilder();
                        char[] buffer = new char[8192];
                        String content = "";
                        int read;
                        while ((read = sr.read(buffer, 0, buffer.length)) != -1) {
                            clientQueue.add(new String(buffer));
                        }
                    } catch (IOException e) {
                        Log.e("ServerReader", "Error in testReading Thread.", e);
                    }
                }
            }
        };

        private Runnable ServerWriter = new Runnable() {
            @Override
            public void run() {
                if (socket != null && socket.isConnected()) {
                    try (OutputStream os = socket.getOutputStream()) {
                        while (clientQueue != null && !clientQueue.isEmpty()) {
                            String xml = clientQueue.poll();
                            os.write(xml.getBytes());
                            os.flush();
                        }
                    } catch (IOException e) {
                        Log.e("ServerWriter", "Error in testReading Thread.", e);
                    }
                }
            }
        };

        @Override
        public void run() {
            if (running) { //Service will keep spinning unti the testService ends the loop
                try (Socket socket = new Socket(serverAddress.getHost(), serverAddress.getPort())) {
                    socket.setSoTimeout(0);
                    Log.d("SOCKET", String.format("Local test Port: %s. Remote test Port: %s", socket.getLocalPort(), socket.getPort()));
                    this.socket = socket;
                    //Make the Threads
                    final Thread writer = new Thread(ServerWriter);
                    final Thread reader = new Thread(ServerReader);
                    //Start the Threads
                    writer.start();
                    reader.start();
                    //Join on the Threads so this driver thread will wait until they finish.
                    writer.join();
                    reader.join();
                } catch (UnknownHostException e) {
                    e.printStackTrace();
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

Answer 1

关闭一个套接字的输入或输出流会关闭另一个流和套接字。