多线程图片上传缺少部分图片
异常原因
代码

异常原因
你好！ 由于业务需要，多本书的图片会通过本地服务上传至服务器，在一次大批量上传10本以上书籍的时候发现某几本书缺少几张图片。经过排查 ，发现是多线程的问题。改用线程池后仍然无效。

限制贴上部分代码
代码
加入同步关键字和睡眠机制仍然无效，加入线程池也无效。
  List<Page> pageList = new Vector<>();
            CountDownLatch countDownLatch = new CountDownLatch(fileNameList.size());
            int count = 0;
            for (String fileName : fileNameList) {
                ++count;
                log.info("本次上传的是第" + count + "张图片,图片名称为===> " + fileName);
                uploadExecutor.submit(() -> {
                    BufferedImage bufferedImage = null;
                    try {
                        UploadResult uploadResult = null;
                        for (int i = 0; i < 10; i++) {
                            //上传代码省略
                            if (uploadResult != null) break;
                        }
                        if (uploadResult == null || StringUtils.isEmpty(uploadResult.getUrl()) || uploadResult.getUrl().contains("error")) {
                            log.info("文件" + fileName + "上传失败");
                        }
                        bufferedImage = ImageIO.read(new File(folder + "/" + fileName));
                        int width = bufferedImage.getWidth();
                        int height = bufferedImage.getHeight();
                        Page page = new Page();
						省略

                        page.setPageName(fileName);
                        log.info("图片==> " + fileName + " 成功上传,成功地址为==> " + page.getPageUrl());
                        pageList.add(page);
                    } catch (Exception e) {
                        log.info("上传失败,原因为===》" + e.getMessage());
                        log.error("上传图片异常:" + e.getMessage());
                    } finally {
                        if (bufferedImage != null) {
                            bufferedImage.flush();
                        }
                        countDownLatch.countDown();

                    }
                });
                try {
                    //休眠50毫秒
                    Thread.sleep(50L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            try {
                countDownLatch.await();
            } catch (InterruptedException e) {
                log.error("锁失败");
            }

最后排查出来是这一行代码的问题

bufferedImage = ImageIO.read(new File(folder + “/” + fileName));

这个ImageIO类是线程不安全的，过了这一行值就乱了。

最后加入判断，如果上传成功的数量和需要上传的数量不符合，线程就睡眠一秒，继续重试。

最后终于解决了，真是天坑啊。

但是任然不知道这个IO为啥不报错，异常处理的日志没有打印出来

一：客户端
   public class TCPClientTU {
            public static void main(String[] args) throws IOException {
                /** 实现TCP图片上传客户端
                 *   实现步骤:
                 *    1. Socket套接字连接服务器
                 *    2. 通过Socket获取字节输出流,写图片
                 *    3. 使用自己的流对象,读取图片数据源
                 *         FileInputStream
                 *    4. 读取图片,使用字节输出流,将图片写到服务器
                 *       采用字节数组进行缓冲
                 *    5. 通过Socket套接字获取字节输入流
                 *       读取服务器发回来的上传成功
                 *    6. 关闭资源
                 */
                Socket socket = new Socket("127.0.0.1",8888);
                //获取字节输出流，图片写到服务器端
                OutputStream outputStream = socket.getOutputStream();
                //创建字节输入流，读取本机上的数据源（图片）
                FileInputStream fileInputStream = new FileInputStream("E:\\七岁的绝密图片\\" +
                        "动漫ACG\\颓废.jpg");//路径写自己想上传的图片的路径
                //开始写字节数组
                int length ;
                byte [] bytes = new byte[1024];
                while((length=fileInputStream.read(bytes))!=-1){
                    outputStream.write(bytes, 0, length);
                }
                //终止
                socket.shutdownOutput();
                //服务器回应上传成功
                InputStream inputStream = socket.getInputStream();
                length = inputStream.read(bytes);
                System.out.println(new String(bytes,0,length));
                fileInputStream.available();
                socket.close();
            }
        }

二：Upload类实现Runnable接口
因为在同一时刻可能不止一个用户上传图片，比如CSDN用户在主页上传自己的头像，就可能存在许多用户同时上传的情况，所有在服务器端加上多线程技术，可以更大程度上模拟当时的情景。还有一个细节，就是图片可能会有重名的情况发生，能、那么将会存在覆盖的情况，所以下面的代码针对这个情况做了处理
 public class Upload implements Runnable{
        private Socket socket;
        public Upload(Socket socket){
            this.socket = socket;
        }
    
        public void run() {
            try{
                InputStream inputStream = socket.getInputStream();
                //将目的文件夹封装到File对象
                File Picture = new File("C:\\Users\\18309\\Desktop\\Picture");
                if(!Picture.exists()){
                    Picture.mkdir();
                }
                //防止文件因为重名而被覆盖，重新定义文件名
                //规则：域名+毫秒值+6位随机数
                String filename = "nyist"+System.currentTimeMillis()
                        +new Random().nextInt(999999)+".jpg";
                //创建字节输入流，将图片写进Picture文件夹中
                FileOutputStream fileOutputStream= new FileOutputStream(Picture+
                        File.separator+filename);
                byte [] bytes = new byte[1024];
                int length;
                while((length = inputStream.read(bytes))!=-1){
                    fileOutputStream.write(bytes, 0,length);
                }
                socket.getOutputStream().write("上传成功".getBytes());
                fileOutputStream.close();
                socket.close();
            }catch (Exception ex){
                System.out.println(ex);
            }
        }
    }

三：服务器端
  //服务器永不停歇
    public class TCPThreadServer {
        public static void main(String[] args) throws IOException {
            ServerSocket serverSocket = new ServerSocket(8888);
            //获取一个客户端就必须要开启一个新的线程
            while(true) {
                Socket socket = serverSocket.accept();
                new Thread(new Upload(socket)).start();//传递的是接口的实现类
            }//出现一个客户端就要开一个线程
        }
    }

运行情况与结果展示：

在学习了TCP协议以后，我们可以做一些图片业务中简单的图片上传功能。

单机版的图片上传太鸡肋，一旦人数过多时，服务就会超时，所以就引入了多线程技术，多个线程处理不同的IP(主机)上传图片的业务。

首先看客户端：

public class uploadpicThreadClient {
    public static void main(String[] args) throws IOException {
        //1、创建客户端socket
        Socket socket = new Socket("192.168.223.1", 9999);

        //2、读取客户端要上传的图片
        FileInputStream fis = new FileInputStream("C:\\0.jpg");

        //3、获取socket输出流，将读到的图片数据发给服务端
        OutputStream outputStream = socket.getOutputStream();
        byte[] buf = new byte[1024];

        int len = 0;
        //使用输出流写出图片数据
        while ((len = fis.read(buf)) != -1) {
            outputStream.write(buf, 0, len);
        }

        // 告诉服务端，客户端数据已发送完毕，服务端停止读取
        socket.shutdownOutput();

        //读取服务端返回的数据
        InputStream inputStream = socket.getInputStream();
        //缓冲区存放数据提高读写速度
        byte[] bufIn = new byte[1024];

        int lenIn = inputStream.read(buf);
        String text = new String(buf, 0, lenIn);
        System.out.println(text);

        //关闭资源
        fis.close();
        socket.close();

    }
}


服务端 ：

服务端一定是一直处于待机状态，当有图片上传需求时，就接受需求。

public class uploadpicThreadServer {
    public static void main(String[] args) throws IOException {
        //创建TCP的socket服务端
        ServerSocket ss =new ServerSocket(9999);
        while (true){
            //获取客户端 
            //阻塞式方法，有任务时就开启，否则一直处于等待状态
            Socket socket = ss.accept();
            new Thread(new UploadTask(socket)).start();
        }
    }
}

线程任务：

获得连接的IP、处理发送过来的图片数据等，这些都是每个线程需要执行的任务，所以需放在线程的run( )方法中执行。

public class UploadTask implements Runnable {
    //规定上传文件的最大大小
    private static final int SIZE = 1024*1024*2;
    private Socket socket;

    public UploadTask(Socket socket) {
        this.socket=socket;
    }
    @Override
    public void run() {
        int count=0;
        String ip = socket.getInetAddress().getHostAddress();
        System.out.println(ip+".........connected");
        try {
            //读取客户端发来的数据
            InputStream inputStream = socket.getInputStream();

            //将读取到的文件存储到一个文件中
            File dir =new File("C:\\pic");
            if (!dir.exists()){
                dir.mkdirs();
            }
            File file =new File(dir,ip+".jpg");
            //如果文件已经存在服务端,就文件名IP在原来的基础上+1
            while (file.exists()){
                file = new File(dir,ip+"("+(++count)+").jpg");
            }
            FileOutputStream fos =new FileOutputStream(file);

            byte[] buf=new byte[1024];

            int len=0;
            while ((len=inputStream.read(buf))!=-1){

                fos.write(buf,0,len);
                if (file.length()>SIZE){
                    System.out.println(ip+"文件体积过大");
                    fos.close();
                    socket.close();

                    System.out.println(ip+"....."+file.delete());
                    return;
                }
            }
            //获取socket输出流，将上传成功字样发给服务端
            OutputStream outputStream = socket.getOutputStream();
            outputStream.write("上传成功".getBytes());
            fos.close();
            socket.close();
        }catch (IOException e){
        }
    }
}


测试

首先启动服务端 uploadpicThreadServe

再启动客户端   uploadpicThreadClient

客户端返回：上传成功



服务端返回连接的IP地址



服务端所指定的文件夹中已有两张图片



 

 

 

 

 

 

 

最近因为项目需要，要大规模的上传照片和mp4。一开始的时候，我是直接使用new Thread创建新线程上传，但是因为上传的量实在太大，而且有时候没有网络，这个就会导致线程的极大量的创建。最后导致内存泄漏。于是 ，只能换另外一种方法了---线程池。

 

项目需求是，要将大文件分解成小文件，然后进行分片上传。这个demo里面只是做了文件切割，分片上传和单个上传。而整个大文件切割的时候，会生成多个分片。这个时候需要多线程进行上传，上传完了之后需要发送合并指令。这就会用到线程里面的

countDownLatch.await();

方法。他会等到多个线程执行完全部任务之后才执行下一行代码。这样完美的解决了之前担心的问题（就是不知道什么时候线程已经全部运行完本次任务的问题）。好了，废话少说。源码撸上

package android.rockchip.update.threadpooltask.upload;

import android.os.Bundle;
import android.os.Handler;
import android.os.Looper;
import android.os.Message;

import java.lang.ref.WeakReference;
import java.util.ArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


public class ThreadPoollExcutorUtil {

    //线程进度回调
    private final static int THREAD_PROGRESS_CODE=100;
    //线程完成
    private final static int THREAD_FINISH_CODE=101;
    //线程被中断
    private final static int THREAD_INTERRUPT_CODE=102;
    //所有线程完成
    private final static int THREAD_ALL_SUCCESS_CODE=103;
    //所有线程执行失败
    private final static int THREAD_ALL_FAILED_CODE=104;
    //handler传递进度值
    private final static String THREAD_FILEPATH="THREAD_FILEPATH";
    //handler传递position值
    private final static String THREAD_POSITION="THREAD_POSITION";
    //线程池核心数
    private int threadCore=3;
    //线程池
    private ExecutorService executor;
    //成功的数量
    int successCount=0;
    //失败的数量
    int failedCount=0;

    private OnUploadListener onUploadListener;
    private UploadHandler handler;

    public ThreadPoollExcutorUtil(){
        init();
    }

    public ThreadPoollExcutorUtil(int threadCore){
        this.threadCore=threadCore;
        init();
    }

    public void setOnUploadListener(OnUploadListener uploadListener){
        this.onUploadListener=uploadListener;
    }

    private void init() {
        handler=new ThreadPoollExcutorUtil.UploadHandler(this);
        successCount=0;
        failedCount=0;
        executor= Executors.newFixedThreadPool(threadCore);
    }

    public void shutDownNow(){
        //终端所有线程
        executor.shutdownNow();
    }

    public void submitAll(final ArrayList<UploadItem> uploadItemArrayList){

        //在这里设置每次的任务值，countDownLatch.await();会一直等待，等到线程处理完全部任务之后，
        // 再跑到下一句代码
        CountDownLatch countDownLatch=new CountDownLatch(uploadItemArrayList.size());
        int i=0;
        //历遍文件
        for(UploadItem uploadItem:uploadItemArrayList){
            final Bundle bundle=new Bundle();
            bundle.putString(THREAD_FILEPATH,uploadItem.getFilePath());
            i++;
            executor.submit(new UpLoadFile(countDownLatch, i, uploadItem, new OnThreadResultListener() {
                @Override
                public void onProgressChange(int persent) {

                }

                @Override
                public void onFinish() {
                    Message.obtain(handler,THREAD_FINISH_CODE,bundle).sendToTarget();
                    successCount++;
                    if (successCount==uploadItemArrayList.size()){
                        successCount=0;
                        failedCount=0;
//                        handler.sendEmptyMessage(THREAD_ALL_SUCCESS_CODE);
                    }else if (failedCount!=0 && (failedCount+successCount==uploadItemArrayList.size())){
                        successCount=0;
                        failedCount=0;
//                        handler.sendEmptyMessage(THREAD_ALL_FAILED_CODE);
                    }
                }

                @Override
                public void onInterrupted() {
                    failedCount++;
                    if (successCount+failedCount==uploadItemArrayList.size()){
                        successCount=0;
                        failedCount=0;
//                        handler.sendEmptyMessage(THREAD_ALL_FAILED_CODE);
                    }
                }
            }));
        }

        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //全部任务跑完,开始下一个
        handler.sendEmptyMessage(THREAD_ALL_SUCCESS_CODE);
    }

    private static class UploadHandler extends Handler {

        private WeakReference<ThreadPoollExcutorUtil> weakReference;
        private UploadHandler(ThreadPoollExcutorUtil object){
            super(Looper.getMainLooper());//执行在UI线程
            weakReference=new WeakReference<>(object);
        }

        @Override
        public void handleMessage(Message msg) {
            ThreadPoollExcutorUtil threadPoollExcutorUtil=weakReference.get();
            if (threadPoollExcutorUtil!=null){
                Bundle data= (Bundle) msg.obj;
                String filePath;
                switch (msg.what){
                    case THREAD_PROGRESS_CODE:{
//                        filePath=data.getString(THREAD_FILEPATH);
                        break;
                    }
                    case THREAD_FINISH_CODE:{
                        filePath=data.getString(THREAD_FILEPATH);
                        threadPoollExcutorUtil.onUploadListener.onThreadFinish(filePath);
                        break;
                    }
                    case THREAD_INTERRUPT_CODE:{
                        filePath=data.getString(THREAD_FILEPATH);
//                        threadPoollExcutorUtil.onUploadListener.onAllFailed();
                        break;
                    }
                    case THREAD_ALL_SUCCESS_CODE:{
                        threadPoollExcutorUtil.onUploadListener.onAllSuccess();
                        break;
                    }
                    case THREAD_ALL_FAILED_CODE:{
                        threadPoollExcutorUtil.onUploadListener.onAllFailed();
                        break;
                    }
                }
            }

            super.handleMessage(msg);
        }
    }
}


demo的下载地址：https://download.csdn.net/download/xiaoyifeishuang1/12103605