看到这个标题，相信不少人会感到疑惑，回忆你们自己的场景会发现，在Spring的项目中很少有使用多线程处理任务的，没错，大多数时候我们都是使用Spring MVC开发的web项目，默认的Controller，Service，Dao组件的作用域都是单实例，无状态，然后被并发多线程调用，那么如果我想使用多线程处理任务，该如何做呢？

比如如下场景：

使用spring-boot开发一个监控的项目，每个被监控的业务（可能是一个数据库表或者是一个pid进程）都会单独运行在一个线程中，有自己配置的参数，总结起来就是：

（1）多实例（多个业务，每个业务相互隔离互不影响）

（2）有状态（每个业务，都有自己的配置参数）

如果是非spring-boot项目，实现起来可能会相对简单点，直接new多线程启动，然后传入不同的参数类即可，在spring的项目中，由于Bean对象是spring容器管理的，你直接new出来的对象是没法使用的，就算你能new成功，但是bean里面依赖的其他组件比如Dao，是没法初始化的，因为你饶过了spring，默认的spring初始化一个类时，其相关依赖的组件都会被初始化，但是自己new出来的类，是不具备这种功能的，所以我们需要通过spring来获取我们自己的线程类，那么如何通过spring获取类实例呢，需要定义如下的一个类来获取SpringContext上下文：
/**
 * Created by Administrator on 2016/8/18.
 * 设置Sping的上下文
 */
@Component
public class ApplicationContextProvider implements ApplicationContextAware {

    private static ApplicationContext context;

    private ApplicationContextProvider(){}

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

    public  static <T> T getBean(String name,Class<T> aClass){
        return context.getBean(name,aClass);
    }


}

然后定义我们的自己的线程类，注意此类是原型作用域，不能是默认的单例：
@Component("mTask")
@Scope("prototype")
public class MoniotrTask extends Thread {

    final static Logger logger= LoggerFactory.getLogger(MoniotrTask.class);
    //参数封装
    private Monitor monitor;
    
    public void setMonitor(Monitor monitor) {
        this.monitor = monitor;
    }

    @Resource(name = "greaterDaoImpl")
    private RuleDao greaterDaoImpl;

    @Override
    public void run() {
        logger.info("线程:"+Thread.currentThread().getName()+"运行中.....");
    }

}

写个测试例子，测试下使用SpringContext获取Bean，查看是否是多实例：
/**
 * Created by Administrator on 2016/8/18.
 */
@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest(classes =ApplicationMain.class)
public class SpingContextTest {



    @Test
    public void show()throws Exception{
        MoniotrTask m1=   ApplicationContextProvider.getBean("mTask", MoniotrTask.class);
        MoniotrTask m2=ApplicationContextProvider.getBean("mTask", MoniotrTask.class);
        MoniotrTask m3=ApplicationContextProvider.getBean("mTask", MoniotrTask.class);
        System.out.println(m1+" => "+m1.greaterDaoImpl);
        System.out.println(m2+" => "+m2.greaterDaoImpl);
        System.out.println(m3+" => "+m3.greaterDaoImpl);

    }


}

运行结果如下：
[ INFO ] [2016-08-25 17:36:34] com.test.tools.SpingContextTest [57] - Started SpingContextTest in 2.902 seconds (JVM running for 4.196)
2016-08-25 17:36:34.842  INFO 8312 --- [           main] com.test.tools.SpingContextTest          : Started SpingContextTest in 2.902 seconds (JVM running for 4.196)
Thread[Thread-2,5,main] => com.xuele.bigdata.xalert.dao.rule.impl.GreaterDaoImpl@285f38f6
Thread[Thread-3,5,main] => com.xuele.bigdata.xalert.dao.rule.impl.GreaterDaoImpl@285f38f6
Thread[Thread-4,5,main] => com.xuele.bigdata.xalert.dao.rule.impl.GreaterDaoImpl@285f38f6

可以看到我们的监控类是多实例的，它里面的Dao是单实例的，这样以来我们就可以在spring中使用多线程处理我们的任务了。

如何启动我们的多线程任务类，可以专门定义一个组件类启动也可以在启动Spring的main方法中启动，下面看下，如何定义组件启动：
@Component
public class StartTask   {

    final static Logger logger= LoggerFactory.getLogger(StartTask.class);
    
    //定义在构造方法完毕后，执行这个初始化方法
    @PostConstruct
    public  void init(){

        final List<Monitor> list = ParseRuleUtils.parseRules();
        logger.info("监控任务的总Task数：{}",list.size());
        for(int i=0;i<list.size();i++) {
            MoniotrTask moniotrTask=   ApplicationContextProvider.getBean("mTask", MoniotrTask.class);
            moniotrTask.setMonitor(list.get(i));
            moniotrTask.start();
            logger.info("第{}个监控task: {}启动 !",(i+1),list.get(i).getName());
        }

    }


}

最后备忘下logback.xml，里面可以配置相对和绝对的日志文件路径：
<!-- Logback configuration. See http://logback.qos.ch/manual/index.html -->
<configuration scan="true" scanPeriod="10 seconds">
  <!-- Simple file output -->
  <appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
    <!--<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">-->
    <!-- encoder defaults to ch.qos.logback.classic.encoder.PatternLayoutEncoder -->
    <encoder>
        <pattern>
            [ %-5level] [%date{yyyy-MM-dd HH:mm:ss}] %logger{96} [%line] - %msg%n
        </pattern>
        <charset>UTF-8</charset> <!-- 此处设置字符集 -->
    </encoder>

    <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
      <!-- rollover daily 配置日志所生成的目录以及生成文件名的规则，默认是相对路径 -->
      <fileNamePattern>logs/xalert-%d{yyyy-MM-dd}.%i.log</fileNamePattern>
        <!--<property name="logDir" value="E:/testlog" />-->
        <!--绝对路径定义-->
      <!--<fileNamePattern>${logDir}/logs/xalert-%d{yyyy-MM-dd}.%i.log</fileNamePattern>-->
      <timeBasedFileNamingAndTriggeringPolicy
          class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
        <!-- or whenever the file size reaches 64 MB -->
        <maxFileSize>64 MB</maxFileSize>
      </timeBasedFileNamingAndTriggeringPolicy>
    </rollingPolicy>


    <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
      <level>DEBUG</level>
    </filter>
    <!-- Safely log to the same file from multiple JVMs. Degrades performance! -->
    <prudent>true</prudent>
  </appender>


  <!-- Console output -->
  <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
    <!-- encoder defaults to ch.qos.logback.classic.encoder.PatternLayoutEncoder -->
      <encoder>
          <pattern>
              [ %-5level] [%date{yyyy-MM-dd HH:mm:ss}] %logger{96} [%line] - %msg%n
          </pattern>
          <charset>UTF-8</charset> <!-- 此处设置字符集 -->
      </encoder>
    <!-- Only log level WARN and above -->
    <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
      <level>INFO</level>
    </filter>
  </appender>


  <!-- Enable FILE and STDOUT appenders for all log messages.
       By default, only log at level INFO and above. -->
  <root level="INFO">
      <appender-ref ref="STDOUT" />
      <appender-ref ref="FILE" />

  </root>

  <!-- For loggers in the these namespaces, log at all levels. -->
  <logger name="pedestal" level="ALL" />
  <logger name="hammock-cafe" level="ALL" />
  <logger name="user" level="ALL" />
    <include resource="org/springframework/boot/logging/logback/base.xml"/>
    <jmxConfigurator/>
</configuration
有什么问题可以扫码关注微信公众号：我是攻城师(woshigcs)，在后台留言咨询。 技术债不能欠，健康债更不能欠， 求道之路，与君同行。 

最近用到使用多线程处理给用户发送站内消息的问题，想到使用java自带的线程池进行处理这个问题，具体如下： 

定义一个线程:



package com.qlwb.util;

import org.apache.log4j.Logger;

import redis.clients.jedis.ShardedJedis;

import com.qlwb.common.redispool.BusinessRedisParam;
import com.qlwb.common.redispool.BusinessRedisPool;
/**
 * 
 * 
 * @类编号:
 * @类名称:DealUserTools
 * @内容摘要: 消息处理
 * @author:鹿伟伟
 * @创建日期:2016年6月18日 下午1:23:03
 * @修改人:
 * @修改日期:
 * @修改描述:简单描述修改的内容
 * @version 1.0.0
 *
 */
public class DealUserTools extends Thread{
    private final Logger logger = Logger.getLogger(DealUserTools.class);
    private int minId=0;
    private int maxId=0;
    public DealUserTools(int minId,int maxId){
        this.maxId=maxId;
        this.minId=minId;
    }
    public void run(){
        ShardedJedis jedis=null;
        logger.info("站内消息处理开始");
        long t11=System.currentTimeMillis();

        try {
            //jedis=BusinessRedisPool.getInstance().getJedis();
            for(int i=minId;i<=maxId;i++){
                logger.info(Thread.currentThread().getName()+"=="+i);
                //TODO
                //业务处理
                }
            }
        } catch (Exception e) {
            logger.error("站内消息状态设置异常：", e);
        }finally{
            BusinessRedisPool.getInstance().returnResource(jedis);
        }
        long t12=System.currentTimeMillis();
        System.out.println("站内消息处理用时"+(t12-t11)+"ms");
    }
    public static void main(String[] args) {

    }
}


使用线程池



ExecutorService pool = Executors.newFixedThreadPool(count);
        Map<String,Object> mapParam=new HashMap<String, Object>();
        try {

            for(int i=0;i<=count;i++){
                Thread t1= new DealUserTools(minId,maxId);
                pool.execute(t1);
            }
            pool.shutdown();
        } catch (Exception e) {
            logger.error("站内消息getUserId:", e);
        }

日常的批量处理任务中，经常需要使用多线程同时处理大量任务，一次读取一定数量的数据，然后放入线程池中等待线程处理完成，再取一定数量数据进行循环处理。

效率比较低的方式是使用同步的for循环进行处理

其次就是使用多线程处理。一般情况使用多线程都会使用线程池来管理，有些情况下，不能把大量任务一次性丢进线程池中，以为内存有限，一般线程池的阻塞队列也是有界的，超出限制可能OOM或者触发拒绝策略，因此需要分批处理，假设一次性读取5000条数据，则需要先等待线程池处理完这5000条数据再进行下一次处理。这时候我们需要确认开启的多线程中的子任务全部结束，再让主线程去执行下一次处理。

大致总结的几种处理方案代码示例如下，本人水平有限，欢迎各位大佬指点留言，谢谢！

	private static final ForkJoinPool FORK_JOIN_POOL = new ForkJoinPool(2 * CPU_COUNT);
    
    /**1. 使用CountDownLatch*/
    public void testCountDownLatch() {
    	//模拟查询到数据库中待处理数据
        List batchList = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            batchList.add(new java.lang.Object());
        }
        if (CollectionUtils.isEmpty(batchList)) {
            return;
        }
        log.info("开始本次批量处理，数据个数：{}，时间：{}", batchList.size(), LocalDateTime.now());
        final CountDownLatch countDownLatch = new CountDownLatch(batchList.size());
        batchList.forEach(Object -> FORK_JOIN_POOL.execute(() -> {
            try {
                TimeUnit.SECONDS.sleep(new Random().nextInt(10));
                log.info("当前线程休眠完成");
                countDownLatch.countDown();
            } catch (Exception e) {
                log.error("异常", e);
            }
        }));
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log.info("完成本次批量处理，数据个数：{}，时间：{}", batchList.size(), LocalDateTime.now());
    }

    /**2. 使用CyclicBarrier*/
    public void testCyclicBarrier() {
        //模拟查询到数据库中待处理数据
        List<Object> batchList = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            batchList.add(new Object());
        }
        if (CollectionUtils.isEmpty(batchList)) {
            return;
        }
        log.info("开始本次批量处理，数据个数：{}，时间：{}", batchList.size(), LocalDateTime.now());
        final CyclicBarrier cyclicBarrier = new CyclicBarrier(batchList.size(), () -> {
            log.info("完成本次批量处理，数据个数：{}，时间：{}", batchList.size(), LocalDateTime.now());
            testCyclicBarrier();
        });
        batchList.forEach(Object -> FORK_JOIN_POOL.execute(() -> {
            try {
                TimeUnit.SECONDS.sleep(new Random().nextInt(10));
                log.info("当前线程休眠完成");
                cyclicBarrier.await();
            } catch (Throwable e) {
                log.error("异常", e);
            }
        }));
    }

	/**3. 使用CompletionService*/
	public void testCompletionService() {

        for (int j = 0; j < 3; j++) {
            List<Object> batchList = new ArrayList<>();
            for (int i = 0; i < 10; i++) {
                batchList.add(new Object());
            }
            if (CollectionUtils.isEmpty(batchList)) {
            	return;
        	}
            log.info("开始本次批量处理，数据个数：{}，时间：{}", batchList.size(), LocalDateTime.now());
            CompletionService completionService = new ExecutorCompletionService(FORK_JOIN_POOL);
            batchList.forEach(Object -> {
                completionService.submit(() -> {
                    try {
                        TimeUnit.SECONDS.sleep(new Random().nextInt(10));
                        log.info("当前线程休眠完成");
                    } catch (Throwable e) {
                        log.error("异常", e);
                    }
                    return null;
                });
            });
            batchList.forEach(imgRecord -> {
                try {
                    completionService.take().get();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
            log.info("完成本次批量处理，数据个数：{}，时间：{}", batchList.size(), LocalDateTime.now());
        }
    }
    

    /**4. 使用CompletableFuture*/
    public void testCompletableFuture() {
        for (int j = 0; j < 3; j++) {
            List<Object> batchList = new ArrayList<>();
            for (int i = 0; i < 10; i++) {
                batchList.add(new Object());
            }
            if (CollectionUtils.isEmpty(batchList)) {
            	return;
        	}
            log.info("开始本次批量处理，数据个数：{}，时间：{}", batchList.size(), LocalDateTime.now());
            ArrayList<CompletableFuture<?>> futureList = new ArrayList<>();
            batchList.forEach(Object -> {
                final CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                    try {
                        TimeUnit.SECONDS.sleep(new Random().nextInt(10));
                        log.info("当前线程休眠完成");
                    } catch (Throwable e) {
                        log.error("异常", e);
                    }
                }, FORK_JOIN_POOL);
                futureList.add(future);
            });
            CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0])).join();
            log.info("完成本次批量处理，数据个数：{}，时间：{}", batchList.size(), LocalDateTime.now());
        }
    }

一、线程池创建四种方式

Java通过Executors提供四种线程池，分别为：

newCachedThreadPool创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。

newFixedThreadPool 创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。

newScheduledThreadPool 创建一个定时线程池，支持定时及周期性任务执行。

newSingleThreadExecutor 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。

二、有返回值的多线程

ExecutorService接口继承自Executor，Executor中的execute方法无返回值，ExecutorService接口中的方法有返回值。





三、计数器使用

CountDownLatch也是juc包中的一个类，类似倒计时计数器，创建对象时通过构造方法设置初始值，调用CountDownLatch对象的await()方法则处于等待状态，调用countDown()方法就将计数器减1，当计数到达0时，则所有等待者或单个等待者开始执行。

有了计数器就可以暂时将主线程阻塞，等异步的多线程全部执行完毕并返回结果后，再继续执行主线程。

四、线程安全问题

有了上面线程池跟计数器的基础，现在可以动手写一个多线程处理任务并合并数据的demo了。

大致思路就是：创建一个定长的线程池，长度为10，计数器初始值也设置为10。每执行一次，将计数器减一，并且将执行结果添加到list集合中，最终多线程全部执行完毕后，计数器停止等待 主线程继续往下执行，返回list。

public static List<String> getExecutorService() throws InterruptedException{
		System.out.println("开始执行多线程...");
		long startTime = System.currentTimeMillis();
		List<String> list = new ArrayList<>();//存放返回结果
		CountDownLatch countDownLatch = new CountDownLatch(10);
		ExecutorService executorService = Executors.newFixedThreadPool(10);
		for (int i = 0; i < 10; i++) {
			Runnable runnable = new Runnable(){

				@Override
				public void run() {
					 try {
						Thread.sleep(3000);
                        list.add(UUID.randomUUID().toString());
                        System.out.println("当前线程name : "+Thread.currentThread().getName());
                        countDownLatch.countDown();
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
				
			};
			executorService.execute(runnable);
		}
		countDownLatch.await();
		System.out.println("submit总共cost 时间：" + (System.currentTimeMillis()-startTime)/1000 + "秒");
		executorService.shutdown();
		return list;
	}

 执行结果如下：



十个线程全部工作，但是返回值中却有null值，跟想要的结果有点出入，为啥呢？

原因在于：ArrayList是非线程安全的。ArrayList的add方法中有size++,不是一个原子操作，所以线程不安全。



 五、CopyOnWriteArrayList的用法

part4中提到的问题 解决方案很简单，将ArrayList换成CopyOnWriteArrayList即可。

	public static List<String> getExecutorService() throws InterruptedException{
		System.out.println("开始执行多线程...");
		long startTime = System.currentTimeMillis();
		List<String> list = new CopyOnWriteArrayList<>();//存放返回结果
		CountDownLatch countDownLatch = new CountDownLatch(10);
		ExecutorService executorService = Executors.newFixedThreadPool(10);
		for (int i = 0; i < 10; i++) {
			Runnable runnable = new Runnable(){

				@Override
				public void run() {
					 try {
						Thread.sleep(3000);
                        list.add(UUID.randomUUID().toString());
                        System.out.println("当前线程name : "+Thread.currentThread().getName());
                        countDownLatch.countDown();
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
				
			};
			executorService.execute(runnable);
		}
		countDownLatch.await();
		System.out.println("submit总共cost 时间：" + (System.currentTimeMillis()-startTime)/1000 + "秒");
		executorService.shutdown();
		return list;
	}

 

 CopyOnWriteArrayList是Java并发包中提供的一个并发容器，它是个线程安全且读操作无锁的ArrayList，写操作则通过创建底层数组的新副本来实现，是一种读写分离的并发策略，我们也可以称这种容器为"写时复制器"。

优点：读操作时性能很高，因为不需要任何同步措施，适用于读多写少的并发场景。

缺点：①.每次写操作都要copy原容器，频繁的GC，内存压力大。②.由于读写分离的策略，读到的数据很可能是旧数据。