Slow storage scheme

background

There is a large amount of data in the cache to be written to the database or saved to the storage service , In the process ：

1. The database times out due to multiple opening and closing
2. A large amount of data written in a short time may explode the storage

The first idea is to solve the problem in batches batch Storage , For a total of m The data of , It ensures a certain writing efficiency （ Write once n strip ）, And ensure certain safety （ Break a thing into m / n + 1 One transaction ）. And this solution can also solve problem two , Because it is executed in batches , The length of time is lengthened , In fact, it is an effect of flow peak shaving .

Realization

Ideas have , Implementation is simple , Here the mybatis batchUpdate Batch processing , use CompletableFuture And recursive implementation of multi-threaded execution of transactions .
batchUpdate Refer to this document batchUpdate stay mybatis Application . Here is mainly about CompletableFuture.

CompletableFuture As a Java The multithreaded tool class provided by asynchronous package , It's very powerful , We don't need to configure the thread pool ourselves , Use it directly CompletableFuture.runAsync You can achieve the purpose of multi-threaded execution . It can also be executed asynchronously , Reference resources Official documents You know , Yes CompletableFuture The return value of can call whenComplete Method , Execute the next batch after the execution of the previous batch of tasks ：

public void execAsync(List<List<User>> all, int i) {
    
	if(i < all.size()) {
    
		CompletableFuture<Void> future = CompletableFuture.runAsync(()->userService.batchInsert(all.get(i)));
		if(i + 1 < all.size()) {
    
			future.whenComplete(() -> execAsync(all, i + 1));
		}
	}
}

appendix ：zip and gz Decompression and storage of files

public class FileUtil {
    

    public static InputStream getNewInputStream(InputStream inputStream, String fileName, String subfix) {
    
        return handleInputStream(inputStream, fileName, subfix);
    }

    private static InputStream handleInputStream(InputStream inputStream, String fileName, String subfix) {
    
        Assert.isTrue(inputStream != null, " The input stream is null");
        File file = null;
        FileOutputStream fos = null;
        try {
    
            file = File.createTempFile(fileName, subfix);
            if (!file.exists()) {
    
                file.createNewFile();
            }
            fos = new FileOutputStream(file);
            byte[] bytes = new byte[1024];
            while ((inputStream.read(bytes)) != -1) {
    
                fos.write(bytes);
            }
        } catch (IOException e) {
    
            log.error(" Failed to create temporary file , The reason is ：{}", e.getMessage());
        } finally {
    
            Assert.isTrue(fos != null, " File write stream is null");
            try {
    
                inputStream.close();
                fos.flush();
                fos.close();
            } catch (IOException e) {
    
                e.printStackTrace();
            }
        }
        if (subfix.equals(SubFixConfig.ZIP)) {
    
            return unzipFile(file, fileName);
        }
        if (subfix.equals(SubFixConfig.GZ)) {
    
            return ungzFile(file);
        }
        return null;
    }

    private static InputStream unzipFile(File file, String filename) {
    
        InputStream is = null;
        try {
    
            ZipFile zipFile = new ZipFile(file);
            is = zipFile.getInputStream(zipFile.getEntry(filename));
        } catch (IOException e) {
    
            log.error(" Get decompression zip An error occurred in the post file stream , The reason is ：{}", e.getMessage());
        }
        return is;
    }

    private static InputStream ungzFile(File file) {
    
        GZIPInputStream gzipInputStream = null;
        try {
    
            gzipInputStream = new GZIPInputStream(new FileInputStream(file));
        } catch (IOException e) {
    
            log.error(" Get decompression gz An error occurred in the post file stream , The reason is ：{}", e.getMessage());
        }
        return gzipInputStream;
    }
}