Cara menggunakan kolam benang SpringBoot dan kolam benang Java-javaTutorial-php.cn

Prinsip penggunaan dan pelaksanaan kumpulan benang SpringBoot dan kumpulan benang Java

Gunakan kumpulan benang lalai

Kaedah 1: Melalui @Async Anotasi memanggil

public class AsyncTest {
    @Async
    public void async(String name) throws InterruptedException {
        System.out.println("async" + name + " " + Thread.currentThread().getName());
        Thread.sleep(1000);
    }
}

Salin selepas log masuk

dan anotasi @EnableAsync perlu ditambahkan pada kelas permulaan, jika tidak, ia tidak akan berkuat kuasa.

@SpringBootApplication
//@EnableAsync
public class Test1Application {
   public static void main(String[] args) throws InterruptedException {
      ConfigurableApplicationContext run = SpringApplication.run(Test1Application.class, args);
      AsyncTest bean = run.getBean(AsyncTest.class);
      for(int index = 0; index <= 10; ++index){
         bean.async(String.valueOf(index));
      }
   }
}

Salin selepas log masuk

Kaedah 2: Menyuntik terus ThreadPoolTaskExecutor

Anda tidak perlu menambah @EnableAsyncanotasi pada masa ini

@SpringBootTest
class Test1ApplicationTests {

   @Resource
   ThreadPoolTaskExecutor threadPoolTaskExecutor;

   @Test
   void contextLoads() {
      Runnable runnable = () -> {
         System.out.println(Thread.currentThread().getName());
      };

      for(int index = 0; index <= 10; ++index){
         threadPoolTaskExecutor.submit(runnable);
      }
   }

}

Salin selepas log masuk

Maklumat konfigurasi lalai kolam benang

Konfigurasi kumpulan benang SpringBoot biasa:

spring:
  task:
    execution:
      pool:
        core-size: 8
        max-size: 16                          # 默认是 Integer.MAX_VALUE
        keep-alive: 60s                       # 当线程池中的线程数量大于 corePoolSize 时，如果某线程空闲时间超过keepAliveTime，线程将被终止
        allow-core-thread-timeout: true       # 是否允许核心线程超时,默认true
        queue-capacity: 100                   # 线程队列的大小,默认Integer.MAX_VALUE
      shutdown:
        await-termination: false              # 线程关闭等待
      thread-name-prefix: task-               # 线程名称的前缀

Salin selepas log masuk

Prinsip pelaksanaan kumpulan benang SpringBoot ditakrifkan dalam kelas

TaskExecutionAutoConfiguration Pelaksanaan dalaman kelas ini juga berdasarkan java asli ThreadPoolTaskExecutor kelas. Kaedah ThreadPoolExecutor dipanggil dalam kelas induknya, tetapi dalam kelas induk initializeExecutor() ditakrifkan sebagai RejectedExecutionHandler dan private RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.AbortPolicy(); dihantar ke initialize() melalui kaedah AbortPolicy. initializeExecutor()

Perhatikan bahawa dalam kelas

, nama kacang bagi kelas TaskExecutionAutoConfiguration ialah: ThreadPoolTaskExecutor dan applicationTaskExecutor. taskExecutor

// TaskExecutionAutoConfiguration#applicationTaskExecutor()
@Lazy
@Bean(name = { APPLICATION_TASK_EXECUTOR_BEAN_NAME,
      AsyncAnnotationBeanPostProcessor.DEFAUL
          T_TASK_EXECUTOR_BEAN_NAME })
@ConditionalOnMissingBean(Executor.class)
public ThreadPoolTaskExecutor applicationTaskExecutor(TaskExecutorBuilder builder) {
   return builder.build();
}

Salin selepas log masuk

// ThreadPoolTaskExecutor#initializeExecutor()
@Override
protected ExecutorService initializeExecutor(
      ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {

   BlockingQueue<Runnable> queue = createQueue(this.queueCapacity);

   ThreadPoolExecutor executor;
   if (this.taskDecorator != null) {
      executor = new ThreadPoolExecutor(
            this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,
            queue, threadFactory, rejectedExecutionHandler) {
         @Override
         public void execute(Runnable command) {
            Runnable decorated = taskDecorator.decorate(command);
            if (decorated != command) {
               decoratedTaskMap.put(decorated, command);
            }
            super.execute(decorated);
         }
      };
   }
   else {
      executor = new ThreadPoolExecutor(
            this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,
            queue, threadFactory, rejectedExecutionHandler);

   }

   if (this.allowCoreThreadTimeOut) {
      executor.allowCoreThreadTimeOut(true);
   }

   this.threadPoolExecutor = executor;
   return executor;
}

Salin selepas log masuk

// ExecutorConfigurationSupport#initialize()
public void initialize() {
   if (logger.isInfoEnabled()) {
      logger.info("Initializing ExecutorService" + (this.beanName != null ? " &#39;" + this.beanName + "&#39;" : ""));
   }
   if (!this.threadNamePrefixSet && this.beanName != null) {
      setThreadNamePrefix(this.beanName + "-");
   }
   this.executor = initializeExecutor(this.threadFactory, this.rejectedExecutionHandler);
}

Salin selepas log masuk

Mengatasi kumpulan benang lalai

Mengatasi objek

lalai Jenis pemulangan kacang boleh sama ada taskExecutor atau ThreadPoolTaskExecutor. Executor

@Configuration
public class ThreadPoolConfiguration {

    @Bean("taskExecutor")
    public ThreadPoolTaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        //设置线程池参数信息
        taskExecutor.setCorePoolSize(10);
        taskExecutor.setMaxPoolSize(50);
        taskExecutor.setQueueCapacity(200);
        taskExecutor.setKeepAliveSeconds(60);
        taskExecutor.setThreadNamePrefix("myExecutor--");
        taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
        taskExecutor.setAwaitTerminationSeconds(60);
        //修改拒绝策略为使用当前线程执行
        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //初始化线程池
        taskExecutor.initialize();
        return taskExecutor;
    }
}

Salin selepas log masuk

Uruskan berbilang kumpulan benang

Jika terdapat berbilang kumpulan benang, seperti mentakrifkan kumpulan benang lain

, ralat akan dilaporkan jika dilaksanakan secara langsung. Pada masa ini, anda perlu menentukan nama kacang. taskExecutor2

@Bean("taskExecutor2")
public ThreadPoolTaskExecutor taskExecutor2() {
    ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    //设置线程池参数信息
    taskExecutor.setCorePoolSize(10);
    taskExecutor.setMaxPoolSize(50);
    taskExecutor.setQueueCapacity(200);
    taskExecutor.setKeepAliveSeconds(60);
    taskExecutor.setThreadNamePrefix("myExecutor2--");
    taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
    taskExecutor.setAwaitTerminationSeconds(60);
    //修改拒绝策略为使用当前线程执行
    taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
    //初始化线程池
    taskExecutor.initialize();
    return taskExecutor;
}

Salin selepas log masuk

Apabila merujuk kumpulan benang, anda perlu menukar nama pembolehubah kepada nama kacang, supaya ia akan dicari mengikut nama.

@Resource
ThreadPoolTaskExecutor taskExecutor2;

Salin selepas log masuk

Untuk berbilang benang menggunakan anotasi

, hanya nyatakan nama kacang dalam anotasi. @Async

@Async("taskExecutor2")
    public void async(String name) throws InterruptedException {
        System.out.println("async" + name + " " + Thread.currentThread().getName());
        Thread.sleep(1000);
    }

Salin selepas log masuk

Empat strategi penolakan kumpulan benang

Empat jenis kumpulan benang yang biasa digunakan dalam JAVA

Pembina kelas adalah seperti berikut: ThreadPoolExecutor

public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                          BlockingQueue<Runnable> workQueue) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         Executors.defaultThreadFactory(), defaultHandler);
}

Salin selepas log masuk

newCachedThreadPool

Tiada had pada bilangan maksimum benang (

Jika terdapat lebih banyak benang terbiar daripada yang diperlukan, ia akan dikitar semula, jika tidak, benang sedia ada akan digunakan semula. maximumPoolSize=Integer.MAX_VALUE

new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                              60L, TimeUnit.SECONDS,
                              new SynchronousQueue<Runnable>());

Salin selepas log masuk

newFixedThreadPool

Kumpulan benang panjang tetap, tugasan yang melebihi bilangan utas akan menunggu dalam baris gilir.

return new ThreadPoolExecutor(nThreads, nThreads,
                              0L, TimeUnit.MILLISECONDS,
                              new LinkedBlockingQueue<Runnable>());

Salin selepas log masuk

newScheduledThreadPool

adalah serupa dengan

, tiada had atas bilangan utas, tetapi newCachedThreadPool boleh ditentukan. Pelaksanaan tertunda dan pelaksanaan berkala boleh dicapai. corePoolSize

public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue());
}

Salin selepas log masuk

Pelaksanaan berkala:

ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);
scheduledThreadPool.scheduleAtFixedRate(()->{
   System.out.println("rate");
}, 1, 1, TimeUnit.SECONDS);

Salin selepas log masuk

Pelaksanaan tertunda:

scheduledThreadPool.schedule(()->{
   System.out.println("delay 3 seconds");
}, 3, TimeUnit.SECONDS);

Salin selepas log masuk

NewSingleThreadExecutor

kumpulan urutan benang tunggal, yang boleh merealisasikan rangkaian urutan benang tunggal, yang boleh merealisasikan urutan urutan benang tunggal.

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}

Salin selepas log masuk

Empat strategi penolakan dalam kumpulan benang Java

: Kumpulan benang membenarkan pemanggil melaksanakan. CallerRunsPolicy
: Jika kumpulan benang menolak tugasan, ralat akan dilaporkan secara langsung. AbortPolicy
: Jika kumpulan benang menolak tugasan, tugas itu akan dibuang terus. DiscardPolicy
: Jika kumpulan benang menolak tugasan, tugasan tertua yang belum dijalankan dalam kumpulan benang akan terus dibuang dan tugasan baharu akan dibariskan. DiscardOldestPolicy

CallerRunsPolicy

melaksanakan kaedah larian terus dalam urutan utama. Kesan

public static class CallerRunsPolicy implements RejectedExecutionHandler {
 
    public CallerRunsPolicy() { }
 
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            r.run();
        }
    }
}

Salin selepas log masuk

adalah serupa dengan:

Runnable thread = ()->{
   System.out.println(Thread.currentThread().getName());
   try {
      Thread.sleep(0);
   } catch (InterruptedException e) {
      throw new RuntimeException(e);
   }
};

thread.run();

Salin selepas log masuk

AbortPolicy

terus membuang

pengecualian dan menunjukkan maklumat tugasan dan maklumat kumpulan benang. , RejectedExecutionException

public static class AbortPolicy implements RejectedExecutionHandler {
 
    public AbortPolicy() { }
 
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        throw new RejectedExecutionException("Task " + r.toString() +
                                             " rejected from " +
                                             e.toString());
    }
}

Salin selepas log masuk

DiscardPolicy

tidak melakukan apa-apa.

public static class DiscardPolicy implements RejectedExecutionHandler {
 
    public DiscardPolicy() { }
 
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    }
}

Salin selepas log masuk

DiscardOldestPolicy

: Alih keluar tugasan tertua daripada baris gilir. e.getQueue().poll()
: Tugasan semasa ditambahkan pada baris gilir. e.execute(r)

public static class DiscardOldestPolicy implements RejectedExecutionHandler {
 
    public DiscardOldestPolicy() { }
 
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            e.getQueue().poll();
            e.execute(r);
        }
    }
}

Salin selepas log masuk

Prinsip penggunaan semula benang Java

Kolam benang

menyimpan objek java, yang dikekalkan dalam java.util.concurrent.ThreadPoolExecutor.Worker. private final HashSet<Worker> workers = new HashSet<Worker>(); ialah baris gilir yang menyimpan tugasan untuk dilaksanakan Apabila tugasan baharu ditambahkan pada kumpulan benang, tugasan itu akan ditambahkan pada baris gilir workQueue. Pelaksanaan workQueue

private final class Worker
    extends AbstractQueuedSynchronizer
    implements Runnable
{
    /**
     * This class will never be serialized, but we provide a
     * serialVersionUID to suppress a javac warning.
     */
    private static final long serialVersionUID = 6138294804551838833L;

    /** Thread this worker is running in.  Null if factory fails. */
    final Thread thread;
    /** Initial task to run.  Possibly null. */
    Runnable firstTask;
    /** Per-thread task counter */
    volatile long completedTasks;

    /**
     * Creates with given first task and thread from ThreadFactory.
     * @param firstTask the first task (null if none)
     */
    Worker(Runnable firstTask) {
        setState(-1); // inhibit interrupts until runWorker
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }

    /** Delegates main run loop to outer runWorker  */
    public void run() {
        runWorker(this);
    }

    // Lock methods
    //
    // The value 0 represents the unlocked state.
    // The value 1 represents the locked state.

    protected boolean isHeldExclusively() {
        return getState() != 0;
    }

    protected boolean tryAcquire(int unused) {
        if (compareAndSetState(0, 1)) {
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        return false;
    }

    protected boolean tryRelease(int unused) {
        setExclusiveOwnerThread(null);
        setState(0);
        return true;
    }

    public void lock()        { acquire(1); }
    public boolean tryLock()  { return tryAcquire(1); }
    public void unlock()      { release(1); }
    public boolean isLocked() { return isHeldExclusively(); }

    void interruptIfStarted() {
        Thread t;
        if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
            try {
                t.interrupt();
            } catch (SecurityException ignore) {
            }
        }
    }
}

Salin selepas log masuk

objek kerja bergantung pada

Berbeza daripada utas yang biasa kami tulis, utas ini berada dalam gelung dan terus memperoleh tugasan baharu daripada baris gilir untuk pelaksanaan. Oleh itu, benang dalam kumpulan benang boleh digunakan semula, bukannya berakhir selepas pelaksanaan seperti benang yang biasa kita gunakan. runWorker()rreeee

Atas ialah kandungan terperinci Cara menggunakan kolam benang SpringBoot dan kolam benang Java. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!