场景：从磁盘某个地方获取文件然后上传到服务器，或者从服务器上传到外部系统

 考虑因素：网络因素，主要是网络带宽（实际环境遇到过，因为服务部署在不同省份）

单线程： 按一秒一张每天传递：24*60*60*N（N为机器数）； 服务端带宽大的情况下配置为多线程

具体QPS设计，需要结合网络情况

另外考虑到windows电脑做成服务启动应用，具体可以参考github的项目：Releases · winsw/winsw · GitHubA wrapper executable that can run any executable as a Windows service, in a permissive license. - Releases · winsw/winswhttps://github.com/winsw/winsw/releases

win+R进入：services.msc 可查看最终部署的应用

具体操作见最后面

设计思路：队列+多线程

其他方式：redis队列或者mq方式实现，主体思路消费队列

需要注意的是：生产者生产的速度控制，在生产和消费之间找到一个对应的平衡点

适用于一个生产者多个消费者模式，主要是在消费瓶颈上

最终实现思路，文章代码待完善（阉割版），实际项目中已实现设计图内容

临时map的作用是把上传文件后将本地文件删除，避免删除不掉本地文件的情况（此处考虑过一段时间得到最终处理方案）

监控中心作用：定期将未从临时map中清除的文件清理掉，防止意外情况，正确情况下不会存在此情况发生

另外：实际项目中，本地到远程服务再到远程服务，中间搭建了一个中继服务

具体可参考模式：本地项目-----中继服务------远程服务器

maven依赖

<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
    <version>1.18.12</version>
    <scope>provided</scope>
</dependency>

工程结构图

1.工具类获取bean

/**
 * description: 获取bean
 */
@Component
public class SpringBeanUtils implements ApplicationContextAware {

    private static ApplicationContext applicationContext;

    @Override
    public void setApplicationContext(ApplicationContext context) throws BeansException {
        applicationContext = context;
    }

    /**
     * 通过beanName获取bean
     * @param beanName
     * @return
     */
    public static Object getBean(String beanName){
        return applicationContext.getBean(beanName);
    }

    public static ApplicationContext getApplicationContext() {
        return applicationContext;
    }
}

2.静态数据中心

/**
 * description:
 */
@Component
public class MonitorThread {


    @Value("${parent_file_path}")
    private String parent_file_path;

    @Value("${max_thread}")
    private int max_thread;

    @PostConstruct
    public void init() {
        MiddleUtils.setParent_file_path(parent_file_path);
        MiddleUtils.setMax_thread(max_thread);
        new ImageProducer().start();
        new ImageMonitorThread().start();
    }

}

@Component
public class MiddleUtils {

    private static volatile ConcurrentLinkedQueue<String> CONSUMER_FILE = new ConcurrentLinkedQueue<>(); //文件消费队列
    private static volatile ConcurrentHashMap<String, Long> FILE_MAP = new ConcurrentHashMap<>(); //计数器

    private static String parent_file_path = null;

    private static int max_thread = 1;

    public static int getMax_thread() {
        return max_thread;
    }

    public static void setMax_thread(int max_thread) {
        MiddleUtils.max_thread = max_thread;
    }

    public static ConcurrentLinkedQueue<String> getConsumerFile() {
        return CONSUMER_FILE;
    }

    public static void setConsumerFile(ConcurrentLinkedQueue<String> consumerFile) {
        CONSUMER_FILE = consumerFile;
    }

    public static ConcurrentHashMap<String, Long> getFileMap() {
        return FILE_MAP;
    }

    public static void setFileMap(ConcurrentHashMap<String, Long> fileMap) {
        FILE_MAP = fileMap;
    }

    public static String getParent_file_path() {
        return parent_file_path;
    }

    public static void setParent_file_path(String parent_file_path) {
        MiddleUtils.parent_file_path = parent_file_path;
    }
}

3.线程池

@Component
@Slf4j
public class ImageThreadPool {


    @Value("${max_thread}")
    private int max_thread;

    @Bean("imageThread")
    public ThreadPoolTaskExecutor initImagePool() {
        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
        threadPoolTaskExecutor.setCorePoolSize(max_thread);
        threadPoolTaskExecutor.setMaxPoolSize(i * 2);
        threadPoolTaskExecutor.setQueueCapacity(max_thread * 2);
        threadPoolTaskExecutor.setThreadNamePrefix("image_pool");
        threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
        threadPoolTaskExecutor.initialize();
        return threadPoolTaskExecutor;
    }
}

4.生产者

@Slf4j
public class ImageProducer extends Thread {


    private static ImageService imageService;

    @Override
    public void run() {
        while (true) {
            /**
             * https://www.ip138.com/ascii/  中文路径转为ASCII码
             * 做成windows应用
             * https://github.com/winsw/winsw/releases
             */
            getImage(MiddleUtils.getParent_file_path());
        }

    }


    public void getImage(String filePath) {
        if (Objects.isNull(filePath)) {
            log.info("根路径为空");
            return;
        }
        File file = new File(filePath);
        File[] files = file.listFiles();
        if (files == null || files.length < 1) {
            return;
        }

        for (File f : files) {
            if (f.isDirectory()) {
                getImage(f.getAbsolutePath());
            } else if (f.isFile()) {
                /**
                 * 自旋等待
                 */
                while (MiddleUtils.getFileMap().size() > MiddleUtils.getConsumerFile().size()) {
                    try {
                        TimeUnit.MILLISECONDS.sleep(10);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                /**
                 * 将数据加入到队列中
                 */
                Long aLong = MiddleUtils.getFileMap().get(f.getAbsolutePath());
                if (aLong != null) {
                    return;
                }
                MiddleUtils.getFileMap().put(f.getAbsolutePath(), System.currentTimeMillis());
                if (Objects.isNull(imageService)) {
                    imageService = (ImageService) SpringBeanUtils.getBean("imageService");
                }
                try {
                    TimeUnit.SECONDS.sleep(2);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                imageService.uploadImage(f.getAbsolutePath());
            }
        }
    }
}

5.消费者（可以将图片转成base64字符串，后端做解密操作完成上传服务端）

@Service
@Slf4j
public class ImageService {

    @Async("imageThread")
    public void uploadImage(String filePath) {
        log.info(filePath);
        if (Objects.isNull(filePath)) {
            return;
        }
        try {
            /**
             * 具体业务
             */
            log.info("具体业务逻辑");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            MiddleUtils.getFileMap().remove(filePath);
        }
    }
}

pom文件依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>org.example</groupId>
    <artifactId>image_upload</artifactId>
    <version>1.0-SNAPSHOT</version>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.5.6</version>
    </parent>
    <packaging>pom</packaging>

    <modules>
        <module>image_upload_local</module>
        <module>image_upload_server</module>
    </modules>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.12</version>
            <scope>provided</scope>
        </dependency>
    </dependencies>


</project>

local端

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>image_upload</artifactId>
        <groupId>org.example</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>image_upload_local</artifactId>
    <packaging>jar</packaging>
    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>


    <build>
        <finalName>image_upload_local</finalName>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <mainClass>com.unique.local.ImageUploadLocal</mainClass>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

server端也一样

如果部署windows应用的话可以参考如下做法

1.github上下载对应windows位数的exe文件

Releases · winsw/winsw · GitHub

2.将exe文件名和jar文件名配置一样

原始

 修改后

xml文件配置，其中BASE为当前exe文件所在目录

<service>
  
  <!-- ID of the service. It should be unique across the Windows system-->
  <id>image_upload_local</id>
  <!-- Display name of the service -->
  <name>image_upload_local</name>
  <!-- Service description -->
  <description>image_upload_local upload</description>
  
  <!-- Path to the executable, which should be started -->
  <executable>java</executable>
  <arguments>-jar %BASE%\image_upload_local.jar --server.port=9002</arguments>
  <logpath>%BASE%\logs</logpath>

</service>

安装：image_upload_local.exe install

更新：先停服务 win+r进入服务右键停止服务 

卸载程序：image_upload_local.exe uninstall

C:\Users\haoha\Desktop\新建文件夹 (2)\新建文件夹>image_upload_local.exe install
2021-11-13 21:19:20,382 INFO  - Installing service 'image_upload_local (image_upload_local)'...
2021-11-13 21:19:20,406 INFO  - Service 'image_upload_local (image_upload_local)' was installed successfully.

C:\Users\haoha\Desktop\新建文件夹 (2)\新建文件夹>image_upload_local.exe uninstall
2021-11-13 21:21:37,650 INFO  - Uninstalling service 'image_upload_local (image_upload_local)'...
2021-11-13 21:21:37,655 INFO  - Service 'image_upload_local (image_upload_local)' was uninstalled successfully.

C:\Users\haoha\Desktop\新建文件夹 (2)\新建文件夹>

单线程上传

服务端：

  public static void main(String[] args) {
        try ( // 创建一个ServerSocket监听8080端口的客户端请求
              ServerSocket server = new ServerSocket(8000);
// 使用accept()阻塞当前线程，等待客户端请求
              Socket socket = server.accept();
// 由Socket获得输入流，并创建缓冲输入流
              BufferedInputStream in = new BufferedInputStream(socket.getInputStream());
// 由文件输出流创建缓冲输出流
              FileOutputStream out = new FileOutputStream("C:\\Users\\14172\\Pictures\\m66.png")) {
// 准备一个缓冲区
            byte[] buffer = new byte[1024];
// 首次从Socket读取数据
            int len = in.read(buffer);
            while (len != -1) {
// 写入数据到文件
                out.write(buffer, 0, len);
// 再次从Socket读取数据
                len = in.read(buffer);
            }System.out.println("接收完成！");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

客户端：

/**
 * @Author he
 * @Description 测试Socket上传文件的客户端
 * @Date 8.22


 **/
public class UdpClient {
    public static void main(String[] args) {
        try ( // 向本机的8080端口发出请求
              Socket socket = new Socket("127.0.0.1", 8000);
              // 由Socket获得输出流，并创建缓冲输出流
              BufferedOutputStream out = new BufferedOutputStream(socket.getOutputStream());
              // 创建文件输入流
              FileInputStream fin = new FileInputStream("C:\\Users\\14172\\Pictures\\m11.png");
              // 由文件输入流创建缓冲输入流
              BufferedInputStream in = new BufferedInputStream(fin)) {
// 准备一个缓冲区
            byte[] buffer = new byte[1024];
// 首次读取文件
            int len = in.read(buffer);
            while (len != -1) {
// 数据写入Socket
                out.write(buffer, 0, len);
// 再次读取文件
                len = in.read(buffer);
            }
            System.out.println("上传成功！");
        } catch (
                ConnectException e) {
            System.out.println("服务器未启动！");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


}

查看文件：

可以看到自己的目录下已经出现了C:\Users\14172\Pictures\m66.png

多线程上传

改成多线程的形式：

修改服务端：

public class isStream {

    public static void main(String[] args)throws Exception {

        ServerSocket server = new ServerSocket(8000);// 创建一个ServerSocket监听8080端口的客户端请求
// 使用accept()阻塞当前线程，等待客户端请求

        while (true) //（2）开始在这里暂停等待接收客户端的连接,得到一个端到端的Socket管道
        {
            Socket socket = server.accept();
            new ServerReadThread(socket).start();
        }
    }

class ServerReadThread extends Thread {
    static int number = 0;

    private Socket socket;

    public ServerReadThread(Socket socket) {
        this.socket = socket;
    }
@Override
    public void run(){
try(
// 由Socket获得输入流，并创建缓冲输入流
    BufferedInputStream in = new BufferedInputStream(socket.getInputStream());
// 由文件输出流创建缓冲输出流
    FileOutputStream out = new FileOutputStream("C:\\Users\\14172\\socket\\"+number+".png"))
{
    number++;
// 准备一个缓冲区
        byte[] buffer = new byte[1024];
// 首次从Socket读取数据
        int len = in.read(buffer);
        while (len != -1) {
// 写入数据到文件
            out.write(buffer, 0, len);
// 再次从Socket读取数据
            len = in.read(buffer);
            System.out.println("接收完成！");
        }
    } catch (FileNotFoundException fileNotFoundException) {
    fileNotFoundException.printStackTrace();
} catch (IOException ioException) {
    ioException.printStackTrace();
}

}
}

查看结果：

修改客户端

修改为用户输入图片路径，进行多次上传图片：

public class UdpClient {
    public static void main(String[] args) throws IOException {
        int num = 0;
        Socket socket = new Socket("127.0.0.1", 8000);
        // 由Socket获得输出流，并创建缓冲输出流
        BufferedOutputStream out = new BufferedOutputStream(socket.getOutputStream());
        Scanner scanner = new Scanner(System.in);
        // 创建文件输入流
        while (true) {
            num++;
            try ( // 向本机的8080端口发出请求

                  //System.out.println("输入文件路径：");

                  FileInputStream fin = new FileInputStream(scanner.nextLine());
                  // 由文件输入流创建缓冲输入流
                  BufferedInputStream in = new BufferedInputStream(fin)) {
// 准备一个缓冲区
                byte[] buffer = new byte[1024];
// 首次读取文件
                int len = in.read(buffer);
                while (len != -1) {
// 数据写入Socket
                    out.write(buffer, 0, len);
// 再次读取文件
                    len = in.read(buffer);
                }
                System.out.println("上传成功！");
            } catch (ConnectException e) {
                System.out.println("服务器未启动！");
            } catch (IOException e) {
                e.printStackTrace();
            }
if(num == 5) break;
        }

    }


}

查看目录：

发现一个小问题：已经存在的一样的命名的图片不会覆盖
然后有的图片传过去之后格式会损坏，貌似是jpg转为png命名的时候