概述
从数据库中取出多类大量数据（多个List<Map<K,V>>）,需要对每个List遍历把每条数据封装进对应的对象，而且需要对数据进行校验，需要二重及以上for循环，耗时过长有时候还跑着跑着挂了。

经大佬指点解决办法为使用多线程处理for循环，实现主要为两点：

1.异步，主线程不阻塞
2.得到其他线程处理结果

直接使用回调函数，或者使用Future，CompletableFuture，Guava ListenableFuture都可以。

大概的思路就是这样，正好复习一下线程池，从头对用到的知识进行介绍。
线程池

几乎所有需要异步或并发执行任务的程序都可以使用线程池。
三个好处

降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
提高响应速度。当任务到达时，任务可以不需要等到线程创建就能立即执行。
提高线程的可管理性。线程是稀缺资源，如果无限制地创建，不仅会消耗系统资源，

还会降低系统的稳定性，使用线程池可以进行统一分配、调优和监控。但是，要做到合理利用

线程池，必须对其实现原理了如指掌。


创建线程池：线程池的几个参数
new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime,milliseconds,
runnableTaskQueue, handler);


向线程池提交任务

execute()方法： 用于提交不需要返回值的任务，所以无法判断任务是否被线程池执行成功，execute()方法输入的任务是一个Runnable类的实例

threadsPool.execute(new Runnable() {
	@Override
	public void run() {
	// TODO Auto-generated method stub
	}
	});


submit()方法用于提交需要返回值的任务。线程池会返回一个future类型的对象，通过这个future对象可以判断任务是否执行成功，并且可以通过future的get()方法来获取返回值，get()方法会阻塞当前线程直到任务完成，而使用get（long timeout，TimeUnit unit）方法则会阻塞当前线程一段时间后立即返回，这时候有可能任务没有执行完。

Future<Object> future = executor.submit(harReturnValuetask);
	try {	
			Object s = future.get();
		} catch (InterruptedException e) {
		// 处理中断异常
		} catch (ExecutionException e) {
		// 处理无法执行任务异常		
		} finally {
		// 关闭线程池
		executor.shutdown();
		}


关闭线程池：遍历线程池中的工作线程，然后逐个调用线程的interrupt方法来中断线程，所以无法响应中断的任务

可能永远无法终止

shutdownNow：首先将线程池的状态设置成STOP，然后尝试停止所有的正在执行或暂停任务的线程，并返回等待执行任务的列表。如果任务不一定要执行完，可以调用shutdownNow方法
shutdown：只是将线程池的状态设置成SHUTDOWN状态，然后中断所有没有正在执行任务的线程。通常调用shutdown来关闭线程池。
只要调用了这两个关闭方法中的任意一个，isShutdown方法就会返回true。当所有的任务都已关闭后，才表示线程池关闭成功，这时调用isTerminaed方法会返回true。


合理配置线程池：从以下几个角度考虑

任务性质：CPU密集型任务、IO密集型任务和混合型任务。用不同规模的线程池分开处理。

CPU密集型任务应配置尽可能小的线程，如配置Ncpu+1个线程的线程池。
IO密集型任务线程并不是一直在执行任务，则应配置尽可能多的线程，如2*Ncpu。
混合型的任务，如果可以拆分，将其拆分成一个CPU密集型任务和一个IO密集型任务，只要这两个任务执行的时间相差不是太大，那么分解后执行的吞吐量将高于串行执行的吞吐量。如果这两个任务执行时间相差太大，则没必要进行分解。
可以通过Runtime.getRuntime().availableProcessors()方法获得当前设备的CPU个数。


任务的优先级：高、中和低。优先级不同的任务可以使用优先级队列PriorityBlockingQueue来处理。它可以让优先级高的任务先执行。注意　如果一直有优先级高的任务提交到队列里，那么优先级低的任务可能永远不能执行。
任务的执行时间：长、中和短。执行时间不同的任务可以交给不同规模的线程池来处理，或者可以使用优先级队列，让执行时间短的任务先执行
任务的依赖性：是否依赖其他系统资源，如数据库连接。依赖数据库连接池的任务，因为线程提交SQL后需要等待数据库返回结果，等待的时间越长，则CPU空闲时间就越长，那么线程数应该设置得越大，这样才能更好地利用CPU。
优先队列建议使用有界队列。



Executor框架
Java的线程既是工作单元，也是执行机制。从JDK 5开始，把工作单元与执行机制分离开来。工作单元包括Runnable和Callable，而执行机制由Executor框架提供。

两级调度模型

在上层，Java多线程程序通常把应用分解为若干个任务，然后使用用户级的调度器（Executor框架）将这些任务映射为固定数量的线程；
在底层，操作系统内核将这些线程映射到硬件处理器上。
应用程序通过Executor框架控制上层的调度；而下层的调度由操作系统内核控制，下层的调度不受应用程序的控制)


Executor框架的结构

分成三类

任务。包括被执行任务需要实现的接口：Runnable接口或Callable接口。
任务的执行。包括任务执行机制的核心接口Executor，以及继承自Executor的ExecutorService接口。Executor框架有两个关键类实现了ExecutorService接口（ThreadPoolExecutor和ScheduledThreadPoolExecutor）。

Executor是一个接口，它是Executor框架的基础，它将任务的提交与任务的执行分离开来。
ThreadPoolExecutor

是线程池的核心实现类，用来执行被提交的任务。
主要由下列4个组件构成。(具体上一章线程池讲过了)

corePool：核心线程池的大小。
maximumPool：最大线程池的大小。
BlockingQueue：用来暂时保存任务的工作队列。
RejectedExecutionHandler：当ThreadPoolExecutor已经关闭或ThreadPoolExecutor已经饱和时（达到了最大线程池大小且工作队列已满），execute()方法将要调用的Handler。


通常使用工厂类Executors来创建。Executors可以创建3种类型的ThreadPoolExecutor

FixedThreadPool

称为可重用固定线程数的线程池。适用于为了满足资源管理的需求，而需要限制当前线程数量的应用场景，它适用于负载比较重的服务器。
源码：
public static ExecutorService newFixedThreadPool(int nThreads) {
	return new ThreadPoolExecutor(nThreads, nThreads,
					0L, TimeUnit.MILLISECONDS,
					new LinkedBlockingQueue<Runnable>());
}


corePoolSize和maximumPoolSize都被设置为创建FixedThreadPool时指定的参数nThreads。
当线程池中的线程数大于corePoolSize时，keepAliveTime为多余的空闲线程等待新任务的

最长时间，超过这个时间后多余的线程将被终止。这里把keepAliveTime设置为0L，意味着多余的空闲线程会被立即终止。
FixedThreadPool使用无界队列LinkedBlockingQueue作为线程池的工作队列（队列的容量为

Integer.MAX_VALUE），会有如下影响。

当线程池中的线程数达到corePoolSize后，新任务将在无界队列中等待，因此线程池中

的线程数不会超过corePoolSize。
由于上一条，使用无界队列时maximumPoolSize将是一个无效参数。
由于上两条，使用无界队列时keepAliveTime将是一个无效参数。
由于使用无界队列，运行中的FixedThreadPool（未执行方法shutdown()或shutdownNow()）不会拒绝任务（不会调用

RejectedExecutionHandler.rejectedExecution方法）。




所以FixedThreadPool的运行过程为：

如果当前运行的线程数少于corePoolSize，则创建新线程来执行任务。
在线程池完成预热之后（当前运行的线程数等于corePoolSize），将任务加入

LinkedBlockingQueue。
线程执行完1中的任务后，会在循环中反复从LinkedBlockingQueue获取任务来执行。




SingleThreadExecutor

使用单个worker线程的Executor。适用于需要保证顺序地执行各个任务；并且在任意时间点，不会有多个线程是活动的应用场景。
源码
public static ExecutorService newSingleThreadExecutor() {
	return new FinalizableDelegatedExecutorService
			(new ThreadPoolExecutor(1, 1,
						0L, TimeUnit.MILLISECONDS,
						new LinkedBlockingQueue<Runnable>()));
}


SingleThreadExecutor的corePoolSize和maximumPoolSize被设置为1。其他参数与FixedThreadPool相同。
SingleThreadExecutor使用无界队列LinkedBlockingQueue作为线程池的工作队列（队列的容量为Integer.MAX_VALUE）。SingleThreadExecutor使用无界队列作为工作队列对线程池带来的影响与FixedThreadPool相同。


所以SingleThreadExector的运行过程为：

如果当前运行的线程数少于corePoolSize（即线程池中无运行的线程），则创建一个新线程来执行任务。
在线程池完成预热之后（当前线程池中有一个运行的线程），将任务加入LinkedBlockingQueue。
线程执行完1中的任务后，会在一个无限循环中反复从LinkedBlockingQueue获取任务来执行。




CachedThreadPool

根据需要创建新线程的线程池。
CachedThreadPool是大小无界的线程池，适用于执行很多的短期异步任务的小程序，或者是负载较轻的服务器。
源码
public static ExecutorService newCachedThreadPool() {
	return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
					60L, TimeUnit.SECONDS,
					new SynchronousQueue<Runnable>());
}


CachedThreadPool的corePoolSize被设置为0，即corePool为空；
maximumPoolSize被设置为Integer.MAX_VALUE，即maximumPool是无界的。
keepAliveTime设置为60L，意味着CachedThreadPool中的空闲线程等待新任务的最长时间为60秒，空闲线程超过60秒后将会被终止。
CachedThreadPool使用没有容量的SynchronousQueue作为线程池的工作队列，但CachedThreadPool的maximumPool是无界的。这意味着，如果主线程提交任务的速度高于maximumPool中线程处理任务的速度时，CachedThreadPool会不断创建新线程。极端情况下，CachedThreadPool会因为创建过多线程而耗尽CPU和内存资源。


CachedThreadPool的运行过程为：

主线程execute()方法首先执行SynchronousQueue.offer（Runnable task）。如果当前maximumPool中有空闲线程正在执行SynchronousQueue.poll（keepAliveTime，TimeUnit.NANOSECONDS），那么主线程执行offer操作与空闲线程执行的poll操作配对成功，主线程把任务交给空闲线程执行，execute()方法执行完成；否则执行下面的步骤下一步骤。
当初始maximumPool为空，或者maximumPool中当前没有空闲线程时，将没有线程执行SynchronousQueue.poll（keepAliveTime，TimeUnit.NANOSECONDS）。这种情况下，步骤1）将失败。此时CachedThreadPool会创建一个新线程执行任务，execute()方法执行完成。
上一步骤中新创建的线程将任务执行完后，会执行SynchronousQueue.poll（keepAliveTime，TimeUnit.NANOSECONDS）。这个poll操作会让空闲线程最多在SynchronousQueue中等待60秒钟。如果60秒钟内主线程提交了一个新任务（主线程执行步骤1）），那么这个空闲线程将执行主线程提交的新任务；否则，这个空闲线程将终止。由于空闲60秒的空闲线程会被终止，因此长时间保持空闲的CachedThreadPool不会使用任何资源。






ScheduledThreadPoolExecutor


是一ThreadPoolExecutor个实现类，可以在给定的延迟后运行命令，或者定期执行命令ScheduledThreadPool-Executor比Timer更灵活，功能更强大。Timer对应的是单个后台线程，而ScheduledThreadPoolExecutor可以在构造函数中指定多个对应的后台线程数。


ScheduledThreadPoolExecutor为了实现周期性的执行任务，对ThreadPoolExecutor做了如下的修改。

使用DelayQueue作为任务队列。

DelayQueue是一个无界队列，所以ThreadPoolExecutor的maximumPoolSize在ScheduledThreadPool-Executor中没有什么意义（设置maximumPoolSize的大小没有什么效果）。


获取任务的方式不同（后文会说明）。
执行周期任务后，增加了额外的处理（后文会说明）。



ScheduledThreadPoolExecutor的执行主要分为两部分

提交任务：当调用ScheduledThreadPoolExecutor的scheduleAtFixedRate()方法或者scheduleWithFixedDelay()方法时，会向ScheduledThreadPoolExecutor的DelayQueue添加一个实现了RunnableScheduledFutur接口的ScheduledFutureTask。

ScheduledFutureTask主要包含3个成员变量，如下。

long型成员变量time，表示这个任务将要被执行的具体时间。
long型成员变量sequenceNumber，表示这个任务被添加到ScheduledThreadPoolExecutor中的序号。
long型成员变量period，表示任务执行的间隔周期。
DelayQueue封装了一个PriorityQueue，这个PriorityQueue会对队列中的ScheduledFutureTask进行排序。排序时，time小的排在前面（时间早的任务将被先执行）。如果两个ScheduledFutureTask的time相同，就比较sequenceNumber，sequenceNumber小的排在前面（也就是说，如果两个任务的执行时间相同，那么先提交的任务将被先执行）。




执行任务：线程池中的线程从DelayQueue中获取ScheduledFutureTask，然后执行任务。

周期执行任务分成4个步骤：

DelayQueue中获取已到期的ScheduledFutureTask（DelayQueue.take()）。到期任务

是指ScheduledFutureTask的time大于等于当前时间。
public E take() throws InterruptedException {
	final ReentrantLock lock = this.lock;
	//1.获取Lock
	lock.lockInterruptibly();
	//2.获取周期任务
	try {
		for (;;) {
			E first = q.peek();
			if (first == null) {
				 // 2.1 如果PriorityQueue为空，当前线程到Condition中等待；否则执行下面的2.2
				available.await();　　　　　　　　　　
			} else {
				long delay = first.getDelay(TimeUnit.NANOSECONDS);
				if (delay > 0) {
					//2.2 如果PriorityQueue的头元素的time时间比当前时间大，到Condition中等待到time时间；否则执行下面的2.3
					long tl = available.awaitNanos(delay);
				} else {
					//2.3.1 获取PriorityQueue的头元素
					E x = q.poll();　　　　　　　　　　
					assert x != null;
					//2.3.2 如果PriorityQueue不为空，则唤醒在Condition中等待的所有线程
					if (q.size() != 0)
					available.signalAll();　　　　　　　　
					return x;
				}
			}
		}
	} finally {
		//3 释放Lock
		lock.unlock();　　　　　　　　　　　　　
	}
}


ScheduledThreadPoolExecutor在一个循环中执行步骤2，直到线程从PriorityQueue获取到一

个元素之后（执行2.3.1之后），才会退出无限循环（结束步骤2）。


线程1执行这个ScheduledFutureTask。
线程1修改ScheduledFutureTask的time变量为下次将要被执行的时间。
线程1把这个修改time之后的ScheduledFutureTask放回DelayQueue中（DelayQueue.add()）。
public boolean offer(E e) {
	final ReentrantLock lock = this.lock;
	lock.lock(); //1.获取Lock。
	try {
		//2.添加任务。
		E first = q.peek();
		q.offer(e); // 2.1 向PriorityQueue添加任务。
		if (first == null || e.compareTo(first) < 0)
			available.signalAll(); // 2.2 如果在上面2.1中添加的任务是PriorityQueue的头元素，唤醒在Condition中等待的所有线程。
		return true;
	} finally {
		lock.unlock(); // 3 释放Lock
	}
}









Executors可以创建2种类型的ScheduledThreadPoolExecutor：包含若干个线程的ScheduledThreadPoolExecutor，只包含一个线程的ScheduledThreadPoolExecutor。



ScheduledThreadPoolExecutor：创建固定个数线程的ScheduledThreadPoolExecutor。适用于需要多个后台线程执行周期任务，同时为了满足资源管理的需求而需要限制后台线程的数量的应用场景。
SingleThreadScheduledExecutor：创建单个线程的SingleThreadScheduledExecutor，适用于需要单个后台线程执行周期任务，同时需要保证顺序地执行各个任务的应用场景。


Runnable接口和Callable接口的实现类，都可以被ThreadPoolExecutor或ScheduledThreadPoolExecutor执行


异步计算的结果。包括接口Future和实现Future接口的FutureTask类。

FutureTask

实现了Future接口和Runnable接口，所以FutureTask可以交给Executor执行，也可以由调用线程直接执行（FutureTask.run()）。
根据FutureTask.run()方法被执行的时机，FutureTask可以处于下面3种状态。

未启动。FutureTask.run()方法还没有被执行之前，FutureTask处于未启动状态。当创建一个FutureTask，且没有执行FutureTask.run()方法之前，这个FutureTask处于未启动状态
已启动。FutureTask.run()方法被执行的过程中，FutureTask处于已启动状态。
已完成。FutureTask.run()方法执行完后正常结束，或被取消（FutureTask.cancel（…）），或执行FutureTask.run()方法时抛出异常而异常结束，FutureTask处于已完成状态。


get方法和cancel方法的执行示意图
FutureTask是基于AQS来实现的，声明了一个内部私有的继承于AQS的子类Sync，对FutureTask所有公有方法的调用都会委托给这个内部子类

  本博客主要实现一个python多线程异步调用的demo。

  程序功能简介：调用main_func作为主程序，主程序内部创建两个线程，分别建立线程ID、线程名和线程内部执行延迟时间，两个线程内部分别调用函数print_time打印时间。exitFlag作为标志位，如果为1表示线程不打印时间直接退出。

  结果分析：程序主线程已经执行结束，并不影响其建立的线程执行，所以可以实现异步调用。
程序代码：
import threading
import time

exitFlag = 0

class myThread (threading.Thread):
    def __init__(self, threadID, name, delay):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.delay = delay
    def run(self):
        print ("开始线程：" + self.name)
        print_time(self.name, self.delay, 3)
        print ("退出线程：" + self.name)

def print_time(threadName, delay, times):
    for i in range(times):
        if exitFlag:
            threadName.exit()
        time.sleep(delay)
        print ("%s: %s" % (threadName, time.ctime(time.time())))

def main_func():
    # 创建新线程
    thread1 = myThread(1, "Thread-1", 1)
    thread2 = myThread(2, "Thread-2", 2)
    # 开启新线程
    thread1.start()
    thread2.start()
    return '主程序执行结束！'
if __name__ == "__main__":
    result = main_func()
    print(result)

执行结果

                
CompletableFuture 详解

CompletableFuture类实现了CompletionStage和Future接口。Future是Java 5添加的类，用来描述一个异步计算的结果，但是获取一个结果时方法较少,要么通过轮询isDone，确认完成后，调用get()获取值，要么调用get()设置一个超时时间。但是这个get()方法会阻塞住调用线程，这种阻塞的方式显然和我们的异步编程的初衷相违背。
为了解决这个问题，JDK吸收了guava的设计思想，加入了Future的诸多扩展功能形成了CompletableFuture。
CompletionStage是一个接口，从命名上看得知是一个完成的阶段，它里面的方法也标明是在某个运行阶段得到了结果之后要做的事情。
进行变换

QT多线程异步调用，类似MFC的PostThreadMessage
  
blog.csdn.net/dongfangyu/article/details/5930129
  
 
  
在MFC中，每个界面线程都会有一个消息队列，通过函数PostThreadMessage，线程之间可以互发消息，由于Post的方式是非阻塞的，因此系统可以表现出很好的性能。这种消息机制是多线程之间异步调用的极佳方式。
  
笔者最近学习QT4.5.3的时候，想在QT中找到类似于相似的机制，但是暂时没有发现。于是想，没有就创造它。以下描述我如何使QT拥有MFC的消息机制。
  
MFC的消息机制，其实就是通过每个消息ID对应函数地址来实现的。明白了这个，构造此机制就简单了。首先使用StlPort的hash_map建立消息ID与回调函数地址的映射表，之所以用hash_map，是由于其时间复杂度很低，而且不会随着表项的增多而变慢，但会增加内存使用量，在现在内存较大的年代，这个问题不大。
  
其次，是通过QThread继续，得到一个基类线程，该线程假设命名MsgThd，在该线程中，配备有hash_map，同时再使用StlPort的vector建立一个消息队列。重载QThread的run函数，其中是一个循环，该循环中不停从vector中尝试得到消息，若得到消息ID，则尝试根据此消息ID，从hash_map得到回调函数地址，若回调函数地址有效，则执行该函数。否则，休眠一定时间（比如1ms），然后继续从vector中尝试得到消息，周而复始。此时，凡是从MsgThd继续的线程已经具有消息队列了。
  
再次，如何使用该MsgThd呢？方法是，凡是需要消息队列的线程，需从MsgThd继续，定义消息ID，定义回调函数，然后把它们放入hash_map中。举例，若线程A，B都是从MsgThd继续，若线程A要给线程B发一个异步消息，那么只需要线程A往线程B的vector中放入一个消息即可，若你喜爱的话，这个动作的函数接口可以写成PostThreadMessage。当然，存放消息的vector，须用加锁解锁，因为它极有可能出现线程A与线程B同时访问的情况，在QT中，可以使用互斥量QMutex。
  
以上所说，是实现QT多线程异步调用的第一种方式。
  
 
  
以下所说，第二种方式。这是笔者学习QT几天后发现的J，因为随着对QT的熟知，接触到越来越多的内容之后，便发现QT本身有这样的机制。
  
而这种机制存在于函数QApplication::postEvent中，此函数相当于MFC的PostThreadMessage。网上有大量关于QT的讯息，但我还没有找到对这个问题讲得较清楚的。我以下解释，不是最清楚的，但说完了，大家就会使用了。
  
新建一个类，比如MsgThd，继续自QThread，重写run函数，里面就一个函数exec();据说，exec()是消息机制的触发函数。重写event函数，据说，所有事件（Event）都会经过该函数，类似于MFC的PreTranslateMessage。（两个“据说”，说明说法仅供参考，笔者也是刚接触几天，莫怪笔者误人子弟。等我把QT源代码看完之后，就没有“据说”了，呵呵。）
  
 
  
假设线程A需要给线程B发消息，在线程A中有代码举例如下：
  
QEvent* pEvent = new QEvent((QEvent::Type)1234);
  
QApplication::postEvent( pThread_B, pEvent );
  
其中1234，是自己定义的，只要大于QT的保留值1024即可。就像你在MFC自定义消息的时候，需要大于WM_USER（0x0400）一样。
  
上面两句的意思是说，线程A产生了一个事件，它发给了线程B。
  
 
  
因此线程B中的event函数可能要这样写：
  
bool Thread_B::event(QEvent * pEvent )
  
{
  
   
// 截获住自定义的事件
 
  
   
if ( pEvent->type() == (QEvent::Type)1234 )
 
  
   
{
 
  
   
            std::cout<<"这句话替换成你需要调用的函数"<<std::endl;
 
  
          }
  
   
return MsgThd::event(pEvent);
 
  
}
  
这就是第二种方式，实现QT多线程异常调用。
  
 
  
 
  
可以自訂事件類型，最簡單的方式，是透過QEvent::Type指定事件類型的常數值，在建構QCustomEvent時作為建構引數並透過postEvent()傳送事件，例如：
  
const QEvent::Type MyEvent = (QEvent::Type) 9393;    
...    
QApplication::postEvent(object, new QCustomEvent(MyEvent));
  
自訂事件必須定義事件號碼（Event number），自定義的事件號碼必須大於QEvent::Type的列舉值，通常1024以下的值是保留給Qt預先定義的事件類型來使用。
  
object是事件的接受者，使用 postEvent()方法時，事件必須以new的方式建立，在事件處理完畢之後，會自動將事件delete，postEvent()會將事件放至事件佇列的尾端，然後立即返回。若要強迫Qt馬上處理先前postEvent()排到佇列的事件，則可以使用sendPostedEvents()。     
您可以使用sendEvent()方法，事件會立即送至接受者，sendEvent()方法的事件不會被delete，所以通常建立在堆疊（Stack）區，例如：     
  
CustomEvent event("Event Message");    
QApplication::sendEvent(object, &event);
  
     
自訂的事件類型必須是QEvent的子類別，通常繼承QCustomEvent類別，建立自訂事件類別可以獲得更多的型態安全（Type safety）。      
要處理自訂事件，可以重新定義customEvent()方法，例如：      
  
void CustomWidget::customEvent(QCustomEvent *event) {    
CustomEvent *customEvent = static_cast<CustomEvent *>(event);     
    ....      
}
  
  
或是重新定義event()方法，將自訂事件分派給其它函式或直接在event()中處理，例如：
  
bool CustomWidget::event(QEvent *event) {    
    if (event->type() == MyCustomEventType) {    
        CustomEvent *myEvent = static_cast<CustomEvent *>(event);    
        // 對自訂事件的處理，或呼叫其它函式來處理事件    
        return true;    
    }    
    return QWidget::event(event);    
} 
转载于:https://www.cnblogs.com/pulas/archive/2012/11/19/2777623.html