Cara Spring Boot menggunakan kumpulan benang untuk mengendalikan puluhan ribu fungsi pemasukan data

# Prakata

Semasa saya mengerjakan projek dua hari lalu, saya ingin meningkatkan pengoptimuman prestasi memasukkan jadual Kerana terdapat dua jadual, jadual lama disisipkan dahulu, dan kemudian yang baharu jadual dimasukkan. Ia berharga lebih daripada 10,000 Data agak perlahan

Kemudian saya memikirkan kumpulan benang ThreadPoolExecutor, dan menggunakan projek Spring Boot, anda boleh menggunakan kumpulan benang ThreadPoolTaskExecutor yang disediakan oleh Spring untuk merangkum ThreadPoolExecutor, dan terus menggunakan anotasi untuk mendayakannya

# Langkah untuk menggunakan

Mula-mula buat konfigurasi pool thread dan biarkan Spring Boot memuatkannya untuk menentukan cara mencipta ThreadPoolTaskExecutor . Gunakan dua anotasi @Configuration dan @EnableAsync untuk menyatakan Ini ialah kelas konfigurasi, dan ia ialah kelas konfigurasi bagi kumpulan benang

@Configuration
@EnableAsync
public class ExecutorConfig {
    private static final Logger logger = LoggerFactory.getLogger(ExecutorConfig.class);
    @Value("${async.executor.thread.core_pool_size}")    
    private int corePoolSize;   
    
    @Value("${async.executor.thread.max_pool_size}")    
    private int maxPoolSize;   
    
    @Value("${async.executor.thread.queue_capacity}")  
    private int queueCapacity;   
    
    @Value("${async.executor.thread.name.prefix}")  
    private String namePrefix;
    @Bean(name = "asyncServiceExecutor")    
    public Executor asyncServiceExecutor() {   
        logger.info("start asyncServiceExecutor");    
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); 
    
        //配置核心线程数       
        executor.setCorePoolSize(corePoolSize);   
    
        //配置最大线程数      
        executor.setMaxPoolSize(maxPoolSize);   
    
        //配置队列大小     
        executor.setQueueCapacity(queueCapacity);    
    
        //配置线程池中的线程的名称前缀        
        executor.setThreadNamePrefix(namePrefix);
        // rejection-policy：当pool已经达到max size的时候，如何处理新任务        
        // CALLER_RUNS：不在新线程中执行任务，而是有调用者所在的线程来执行  
         
         executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());     
        //执行初始化      
        executor.initialize();   
        return executor;  
     }
}

@Value dikonfigurasikan oleh saya dalam application.properties konfigurasi dan bebas menentukan

# 异步线程配置
# 配置核心线程数
async.executor.thread.core_pool_size = 5
# 配置最大线程数
async.executor.thread.max_pool_size = 5
# 配置队列大小
async.executor.thread.queue_capacity = 99999
# 配置线程池中的线程的名称前缀
async.executor.thread.name.prefix = async-service-

untuk mencipta antara muka Perkhidmatan, iaitu benang tak segerak Antara muka

public interface AsyncService {   
    /**     
      * 执行异步任务     
      * 可以根据需求，自己加参数拟定，我这里就做个测试演示    
      */   
    void executeAsync();
}

Kelas pelaksanaan

@Service
public class AsyncServiceImpl implements AsyncService {  
    private static final Logger logger = LoggerFactory.getLogger(AsyncServiceImpl.class);
    @Override 
    @Async("asyncServiceExecutor")    
    public void executeAsync() {    
        logger.info("start executeAsync");
        System.out.println("异步线程要做的事情");        
        System.out.println("可以在这里执行批量插入等耗时的事情");
        logger.info("end executeAsync");   
    }
}

menambah anotasi @Async("asyncServiceExecutor" ) kepada kaedah executeAsync() Kaedah asyncServiceExecutor ialah nama kaedah dalam ExecutorConfig.java sebelumnya, menunjukkan bahawa kaedah executeAsync memasuki Kumpulan benang dicipta oleh kaedah asyncServiceExecutor

Langkah seterusnya ialah menyuntik Servis melalui anotasi @Autowired dalam Pengawal atau di suatu tempat

@Autowiredprivate 
AsyncService asyncService;

@GetMapping("/async")
public void async(){  
    asyncService.executeAsync();
}

Pencetakan log

2022- 07-16 22:15:47.655 INFO 10516 --- [async-service-5 ] c.u.d.e.executor.impl.AsyncServiceImpl : mulakan executeAsync
Perkara yang perlu dilakukan oleh utas tak segerak
Anda boleh melakukan sisipan kelompok dan tugasan lain yang memakan masa di sini Perkara
2022-07-16:47:65 INFO 22:65 10516 --- [async-service-5] c.u.d.e.executor.impl.AsyncServiceImpl : tamatkan executeAsync
2022-07-16 22:15:47.770 INFO 10516 --- [async-service-1] c.u.executord.imepl. AsyncServiceImpl : mulakan executeAsync
Apa yang perlu dilakukan oleh benang tak segerak
Perkara yang memakan masa seperti sisipan kelompok boleh dilakukan di sini
2022-07- 16 22:15:47.770 INFO 10516 --- [async service-1] c.u.d.e.executor.impl.AsyncServiceImpl : tamatkan executeAsync
2022-07-16 22:15:47.816 MAKLUMAT 10516 --- [async-service- 2] c.u.d.e.executor startS.impl.Asyncl Perkara yang perlu dilakukan oleh benang tak segerak
Perkara yang memakan masa seperti sisipan kelompok boleh dilakukan di sini
2022-07-16 22:15:47.816 INFO 10516 - -- [async-service-2] c.u.d.e.executor. impl.AsyncServiceImpl : tamatkan executeAsync
16-07-2022 22:15:48.833 MAKLUMAT 10516 --- [async-service-3] c.u.d.e.executor.impl.AsyncServiceImpl : apa yang ada pada asynchronousAsync🎜
Anda boleh melakukan perkara yang memakan masa seperti sisipan kelompok di sini
2022-07-16 22:15:48.834 INFO 10516 --- [async-service-3] c.u.d.e.executor.impl.AsyncServiceImpl 2022-07-16 22:15:48.986 INFO 10516 --- [async-service-4] c.u.d.e.executor.impl.AsyncServiceImpl : mula executeAsync
Asynchronous thread untuk melakukan Perkara yang anda boleh lakukan 2022-07-16 22:15:48.987 INFO 10516 --- [async-service-4] c.u.d.e.executor.impl.AsyncServiceImpl : end executeAsync

F log di atas, kami dapati bahawa [async-service-] mempunyai berbilang benang, yang jelas dilaksanakan dalam kumpulan benang yang kami konfigurasikan, dan dalam setiap permintaan, pengawal mula dan log tamat dicetak secara berterusan, menunjukkan bahawa setiap permintaan telah dijawab dengan cepat, dan operasi yang memakan masa diserahkan kepada benang dalam kumpulan benang untuk pelaksanaan tak segerak

Walaupun kami telah menggunakan kumpulan benang, Masih tidak jelas keadaan kumpulan benang pada masa itu; . Di sini saya mencipta subkelas ThreadPoolTaskExecutor Setiap kali utas diserahkan, status larian kumpulan utas semasa akan dicetak

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.concurrent.ListenableFuture;
import java.util.concurrent.Callable;import java.util.concurrent.Future;import java.util.concurrent.ThreadPoolExecutor;
/** 
* @Author: 腾腾 
* @Date: 2022/7/16/0016 22:19 
*/
public class VisiableThreadPoolTaskExecutor extends ThreadPoolTaskExecutor {
    private static final Logger logger = LoggerFactory.getLogger(VisiableThreadPoolTaskExecutor.class);
    private void showThreadPoolInfo(String prefix) {        
        ThreadPoolExecutor threadPoolExecutor = getThreadPoolExecutor();
        if (null == threadPoolExecutor) {    
            return;  
        }
        logger.info("{}, {},taskCount [{}], completedTaskCount [{}], activeCount [{}], queueSize [{}]",                
        this.getThreadNamePrefix(),         
        prefix,           
        threadPoolExecutor.getTaskCount(),     
        threadPoolExecutor.getCompletedTaskCount(),  
        threadPoolExecutor.getActiveCount(),    
        threadPoolExecutor.getQueue().size());
     }
    @Override    
    public void execute(Runnable task) {    
        showThreadPoolInfo("1. do execute");       
        super.execute(task);   
    }
    @Override    
    public void execute(Runnable task, long startTimeout) {      
        showThreadPoolInfo("2. do execute");   
        super.execute(task, startTimeout);  
    }
    @Override  
    public Future<?> submit(Runnable task) {   
        showThreadPoolInfo("1. do submit");    
        return super.submit(task);   
    }
    @Override  
    public <T> Future<T> submit(Callable<T> task) {   
        showThreadPoolInfo("2. do submit");      
        return super.submit(task); 
    }
    @Override    
    public ListenableFuture<?> submitListenable(Runnable task) {    
        showThreadPoolInfo("1. do submitListenable");   
        return super.submitListenable(task);   
    }
    @Override
    public <T> ListenableFuture<T> submitListenable(Callable<T> task) {     
        showThreadPoolInfo("2. do submitListenable");     
        return super.submitListenable(task);  
    }
}

Seperti yang ditunjukkan di atas, kaedah showThreadPoolInfo akan mencetak jumlah tugas, bilangan tugasan selesai, aktif Bilangan utas dan saiz baris gilir dicetak, dan kemudian laksana, serahkan dan kaedah lain kelas induk ditindih, dan kaedah showThreadPoolInfo dipanggil di dalam, supaya setiap kali tugasan diserahkan kepada kolam benang, keadaan asas kumpulan benang semasa akan dicetak ke log.

Ubah suai kaedah asyncServiceExecutor ExecutorConfig.java dan tukar ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor() kepada ThreadPoolTaskExecutor executor = VisiableThreadPoolTaskExecutor()

2022-07-16 22:23:30.951 INFO 14088 --- [nio-8087-exec-2] u.d.e.e.i.VisiableThreadPoolTaskExecutor: asyncunt-service-, 2. do submit [0]taskCount complete 0], activeCount [0], queueSize [0]
2022-07-16 22:23:30.952 INFO 14088 --- [async-service-1] c.u.d.e.executor.impl.AsyncServiceImpl : mulakan executeAsyncAnda boleh melakukan perkara yang memakan masa seperti sisipan kelompok di sini
2022-07-16 22:23:30.953 INFO 14088 --- [async-service-1] c.u.d.e.executor.impl .AsyncServiceImpl : tamatkan executeAsync
2022-07-16 22:23:31.351 MAKLUMAT 14088 --- [nio-8087-exec-3] u.d.e.e.i.VisiableThreadPoolTaskExecutor-,]async.Co.TaskExecutor do : async.Co. completeTaskCount [ 1], activeCount [0], queueSize [0]
2022-07-16 22:23:31.353 INFO 14088 --- [async-service-2] c.u.d.e.executor.impl.AsyncServiceImpl : start execute >asynchronous Perkara yang perlu dilakukan oleh benang
Anda boleh melakukan perkara yang memakan masa seperti sisipan kelompok di sini
2022-07-16 22:23:31.353 INFO 14088 --- [async-service-2] c.u.d.e.executor .impl.AsyncServiceImpl : tamatkan executeAsync
2022-07-16 22:23:31.927 MAKLUMAT 14088 --- [nio-8087-exec-5] u.d.e.e.i.VisiableThreadPool.Counter-servis [penyerah-penyerah2,Cotasserah-tugas-2 ], completeTaskCount [ 2], activeCount [0], queueSize [0]
2022-07-16 22:23:31.929 INFO 14088 --- [async-service-3] c.u.d.e.executor.impl.AsyncServiceexecuteImpl :
asynchronous Perkara yang perlu dilakukan oleh benang
Anda boleh melakukan perkara yang memakan masa seperti sisipan kelompok di sini
2022-07-16 22:23:31.930 INFO 14088 --- [async-service-3] c.u.d.e.executor.impl.AsyncServiceImpl : tamatkan executeAsync
2022-07-16 22:23:32.496 MAKLUMAT 14088 --- [nio-8087-exec-7] u.d.e.e.i.PoolisiableTsynchexservis unt [3], completeTaskCount [ 3], activeCount [0], queueSize [0]
2022-07-16 22:23:32.498 INFO 14088 --- [async-service-4] c.u.d.e.executor.impl.AsyncServiceImpl : mulakan executeAsync
asynchronous Perkara yang perlu dilakukan oleh benang
Anda boleh melakukan perkara yang memakan masa seperti sisipan kelompok di sini
2022-07-16 22:23:32.499 INFO 14088 --- [async-service- 4] c.u.d.e.executor.impl.AsyncServiceImpl : tamatkan executeAsync

Perhatikan baris log ini:

2022-07-16 22:23:32.496 INFO [nio-8087-exec-7] u.d.e.e.i.VisiableThreadPoolTaskExecutor: async-service-, 2. serahkan,taskCount [3], completeTaskCount [3], activeCount [0], queueSize [0]

Atas ialah kandungan terperinci Cara Spring Boot menggunakan kumpulan benang untuk mengendalikan puluhan ribu fungsi pemasukan data. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!