由 dawei 副标题#e#
一个线程从某个地方接收消息(数据)，可以是其他主机或者消息队列，然后转由另外的一个线程池来执行具体处理消息的逻辑，并且消息的处理速度小于接收消息的速度。这种情景很常见，试想一下，你会怎么设计和实现？
直观想法
很显然采用JUC的线程框架，可以迅速写出代码。
消息接收者：
public class Receiver {
private static volatile boolean inited = false;
private static volatile boolean shutdown = false;
private static volatile int cnt = 0;
private MessageHandler messageHandler;
public void start(){
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
while(!shutdown){
init();
recv();
}
}
});
}
/**
* 模拟消息接收
*/
public void recv(){
Message msg = new Message("Msg" + System.currentTimeMillis()); System.out.println(String.format("接收到消息(%d): %s", ++cnt, msg)); messageHandler.handle(msg); } public void init(){ if(!inited){ messageHandler = new MessageHandler(); inited = true; } } public static void main(String[] args) { new Receiver().start();
}
}
消息处理：
public class MessageHandler {
private static final int THREAD_POOL_SIZE = 4;
private ExecutorService service = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
public void handle(Message msg) {
try {
service.execute(new Runnable() {
@Override
#p#副标题#e#
public void run() {
parseMsg(msg);
}
});
} catch (Throwable e) {
System.out.println("消息处理异常" + e); } } /** * 比较耗时的消息处理流程 */ public void parseMsg(Message message) { while (true) { try { System.out.println("解析消息：" + message); Thread.sleep(5000); System.out.println("============================"); } catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
#p#副标题#e##p#分页标题#e#
效果：这种方案导致的现象是接收到的消息会迅速堆积，我们从消息队列（或者其他地方）取出了大量消息，但是处理线程的速度又跟不上，所以导致的问题是大量的Task会堆积在线程池底层维护的一个阻塞队列中，这会极大的耗费存储空间，影响系统的性能。
分析：当execute()一个任务的时候，如果有空闲的worker线程，那么投入运行，否则看设置的***线程个数，没有达到线程个数限制就创建新线程，接新任务，否则就把任务缓冲到一个阻塞队列中，问题就是这个队列，默认的大小是没有限制的，所以就会大量的堆积任务，必然耗费heap空间。