多线程整理

1.建立线程池

package cn.cw.study.config;

import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

@EnableAsync
@Configuration
public class ThreadPoolConfig {
    Logger logger = LoggerFactory.getLogger(ThreadPoolConfig.class);

    // 核心线程池大小
    private int corePoolSize = 20;

    // 最大可创建的线程数
    private int maxPoolSize = 100;

    // 队列最大长度
    private int queueCapacity = 500;

    // 线程池维护线程所允许的空闲时间
    private int keepAliveSeconds = 30;


    // ThredPoolTaskExcutor的处理流程
    // 当池子大小小于corePoolSize，就新建线程，并处理请求
    // 当池子大小等于corePoolSize，把请求放入workQueue中，池子里的空闲线程就去workQueue中取任务并处理
    // 当workQueue放不下任务时，就新建线程入池，并处理请求，如果池子大小撑到了maximumPoolSize，就用RejectedExecutionHandler来做拒绝处理
    // 当池子的线程数大于corePoolSize时，多余的线程会等待keepAliveTime长时间，如果无请求可处理就自行销毁
    @Bean("getAsyncExecutor")
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor =new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(corePoolSize);
        executor.setMaxPoolSize(maxPoolSize);
        executor.setQueueCapacity(queueCapacity);
        executor.setKeepAliveSeconds(keepAliveSeconds);
        executor.setThreadNamePrefix("thread-pool-");
        // 线程池对拒绝任务(无线程可用)的处理策略
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        logger.debug("----------------已初始化线程池-------------------");
        return executor;
    }
}

2.异步方法调用

package cn.cw.study.config;

import java.util.concurrent.Future;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Service;

import cn.cw.study.bean.TestTable;
import cn.cw.study.mapper.TestTableMapper;

@Service
public class TaskService {

    @Autowired
    TestTableMapper testTableMapper;
    /**
     * 方法异步调用，无返回值
     * @param i
     */
    @Async
    public void insertData(int number) {
        System.out.println(Thread.currentThread().getName()+"异步处理-------"+number);
    }

    /**
     * 异步方法调用，有返回值
     * @param number
     * @return
     */
    @Async
    public Future<String> executeFutureData(int number){

        Future<String> future = new AsyncResult<String>("success is-----"+number);

        TestTable table =new TestTable();
        table.setId("test"+number);
        table.setIp("123123213213213213");
        table.setName("test"+number);
        table.setPassword("test_password");

        int createTable = testTableMapper.insert(table);

        System.out.println(Thread.currentThread().getName()+"异步处理有返回值------"+table.getId());
        return future;
    }
}

3.测试

package cn.cw.study.config.util;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import org.apache.ibatis.javassist.expr.NewArray;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.core.task.TaskRejectedException;
import org.springframework.test.context.junit4.SpringRunner;

import cn.cw.study.bean.TestTable;
import cn.cw.study.config.TaskService;
import cn.cw.study.mapper.TestTableMapper;

@RunWith(SpringRunner.class)
@SpringBootTest
public class TestTask {

    @Autowired
    TaskService taskService;

    @Autowired
    TestTableMapper testTableMapper;


//    @Test
    public void testvoid() {
        for(int i=0;i<2000;i++) {
            taskService.insertData(i);
        }
    }


    @Test
    public void testResult() throws InterruptedException, ExecutionException {
        List<Future<String>> list = new ArrayList<Future<String>>();
        long start = System.currentTimeMillis();
        for(int j=0;j<6;j++) {
            for(int i=0;i<400;i++) {
                while(true) {
                    try {
                        Future<String> executeFutureData = taskService.executeFutureData(Integer.parseInt(j+""+i));
                        list.add(executeFutureData);
                        break;
                    } catch (TaskRejectedException  e) {
                        System.out.println("线程池满，等待1S");
                        Thread.sleep(1000);
                    }
                }
            }
            Thread.sleep(5000);
        }

        long end = System.currentTimeMillis();
        System.out.println("代码执行时间"+(end-start));

//        for(Future<String> future:list) {
//            System.out.println(future.get());
//        }


    }

}