Java多线程Concurrency与Parallelism实践：Socket多客户端服务端异常咨询

我在做并发（Concurrency）与并行（Parallelism）概念验证的实践项目时，遇到多客户端接入异常问题，具体内容如下：

项目说明 #

• 共包含4个核心类：
  • 2个客户端类：Agsqr负责生成并发送平方数序列，AgFibo负责生成并发送斐波那契数列
  • 1个服务端类：agclassserver负责接收两类客户端数据，排序后以列表形式展示
  • 1个任务处理类：ClassesHandler是服务端的多线程任务实现，用于处理每个客户端的连接请求
• 预期效果：服务端支持多客户端同时接入，同时接收平方数和斐波那契数列两类数据，排序后输出

异常现象 #

依次启动agclassserver服务端、Agsqr和AgFibo客户端后，服务端仅能识别首个接入的客户端。例如先启动Agsqr接入，后续启动AgFibo接入时，服务端不会输出斐波那契数列，反而重复输出平方数，相当于得到两份Agsqr的运行结果。

相关代码 #

第一版服务端代码（存在核心错误）agclassserver #

package smatp2;

import java.io.DataInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.Collections;
import java.util.concurrent.*;

public class agclassserver {
    
    public static void main(String[] args) throws IOException, InterruptedException {
        ArrayList<Integer> array  = new ArrayList<Integer>();
        ArrayList<ClassesHandler> classes_handler  = new ArrayList<ClassesHandler>();
        ExecutorService service= Executors.newFixedThreadPool(10);
        
        boolean var=true;
        int message;
        
            try {   
                while(true) {
                    ServerSocket server = new ServerSocket(9090);
                    System.out.println("I waiting for the client ");
                    Socket socket = server.accept();// waiting for client to connect with server
                    ClassesHandler client = new ClassesHandler(socket);
                    classes_handler.add(client);
                    service.execute(client);
                }
            }catch (Exception e) {
                
            }   
    }
}

修正版服务端代码 agclassserver #

package smatp2;

import java.io.DataInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.Collections;
import java.util.concurrent.*;

public class agclassserver {
    
    public static void main(String[] args) throws IOException, InterruptedException {
        ArrayList<Integer> array  = new ArrayList<Integer>();
        ArrayList<Integer> arraySQR  = new ArrayList<Integer>();
        ArrayList<Integer> arrayFibo  = new ArrayList<Integer>();
        ArrayList<ClassesHandler> classes_handler  = new ArrayList<ClassesHandler>();
        ExecutorService service= Executors.newFixedThreadPool(10);
        
        boolean var=true;
        int message;
        
            try {   
                ServerSocket server = new ServerSocket(9090);
                while(true) {
                    
                    System.out.println("I waiting for the client ");
                    Socket socket = server.accept();// waiting for client to connect with server
                    String hostName = socket.getInetAddress().getHostName();
                    ClassesHandler client = new ClassesHandler(socket,array,arraySQR,arrayFibo,hostName);
                    classes_handler.add(client);
                    service.execute(client);
                    
                }
            }catch (Exception e) {
                
            }   
    }
}

平方数客户端代码 Agsqr #

package smatp2;

import java.io.DataOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.net.Socket;

public class Agsqr  {
    public static void main(String[] args) throws IOException, InterruptedException {
        try {
        Socket socket = new Socket ("localhost", 9090);
        int n=0;
        OutputStream output = socket.getOutputStream();
        DataOutputStream stream = new DataOutputStream(output);
        while(true) {
            
            int result = n*n;
            stream.writeInt(result);
            n++;
            Thread.sleep(1000);
        }
    }catch(Exception e) {
        
    }
    }
}

斐波那契客户端代码 AgFibo #

package smatp2;

import java.io.DataOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.net.Socket;
import java.net.UnknownHostException;

public class AgFibo extends Thread {
    
    public synchronized void run() {

    }

    public static void main(String[] args) throws UnknownHostException, IOException, InterruptedException {
        try {
             int previousNumber = 0;
             int nextNumber = 1;
             boolean variable=false ;
            Socket socket = new Socket ("localhost", 9090);
            OutputStream output = socket.getOutputStream();
            DataOutputStream stream = new DataOutputStream(output);
            
                while(!variable){
                    stream.writeInt(previousNumber);
                   int sum = previousNumber + nextNumber;
                   previousNumber = nextNumber;
                   nextNumber = sum;    
                   Thread.sleep(1000);
               }
            
        }catch(Exception e) {
            
        }
    }
}

任务处理类代码 ClassesHandler #

package smatp2;

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

public class ClassesHandler implements Runnable {
    private Socket socket;
    private int message;
    ArrayList<Integer> array  = new ArrayList<Integer>();
    
    private DataInputStream dataInputStream;
    public ClassesHandler(Socket socket) throws IOException{
        this.socket = socket;
        dataInputStream = new DataInputStream(socket.getInputStream());
    }
    
    @Override
    public void run() {
        try {
            while(true) {
                message = dataInputStream.readInt();
                array.add(message);
                Collections.sort(array);
                System.out.println(message);
                System.out.println(array);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

问题根因 #

1. 第一版服务端的核心错误：ServerSocket初始化放在了while(true)循环内部，每次循环都会重新创建一个监听9090端口的ServerSocket实例。首个客户端接入后，accept()方法返回，进入下一次循环时旧的ServerSocket会被回收，端口被释放后重新绑定，第二个客户端实际上连接到了新的ServerSocket实例，但前一个客户端的连接也因为旧ServerSocket销毁出现异常，最终导致只能拿到首个客户端的数据。
2. 并发安全问题：原ClassesHandler中每个线程维护独立的ArrayList，无法实现多客户端数据合并统计；且ArrayList是非线程安全的，如果多线程同时操作共享列表会出现并发修改异常。
3. 客户端识别逻辑缺失：当前服务端没有识别客户端类型的逻辑，无法区分收到的数据是平方数还是斐波那契数列，无法实现分类存储。

修复方案 #

1. 将ServerSocket的初始化移到while(true)循环外部，确保整个服务运行期间只创建一个监听9090端口的实例，持续接收客户端连接。
2. 如果需要合并多客户端数据，将共享列表传入ClassesHandler时，使用Collections.synchronizedList()包装列表，或者改用CopyOnWriteArrayList保证线程安全。
3. 客户端首次连接时先发送1位标识：比如Agsqr发送0，AgFibo发送1，服务端ClassesHandler首次读取到标识后，后续数据存入对应类型的列表即可。

内容的提问来源于stack exchange，提问作者zahreddine