Java实现的线程池、消息队列功能方法

忘是亡心i 2022-04-16 02:56 209阅读 0赞

ThreadPoolManager类：负责管理线程池，调用轮询的线程来访问字符串缓冲区的内容，维护缓冲区，当线程池溢出时抛出的Runnable任务被加入到字符缓冲区。  
  
文章来自中国建站：http://www.jz123.cn/text/0127908.html  
  
  
  
package com.abin.message.line;  
  
import java.util.LinkedList;  
import java.util.Queue;  
import java.util.concurrent.ArrayBlockingQueue;  
import java.util.concurrent.Executors;  
import java.util.concurrent.RejectedExecutionHandler;  
import java.util.concurrent.ScheduledExecutorService;  
import java.util.concurrent.ScheduledFuture;  
import java.util.concurrent.ThreadPoolExecutor;  
import java.util.concurrent.TimeUnit;  
  
public class ThreadPoolManager \{  
private static ThreadPoolManager tpm = new ThreadPoolManager();  
// 线程池维护线程的最少数量  
private final static int CORE\_POOL\_SIZE = 4;  
// 线程池维护线程的最大数量  
private final static int MAX\_POOL\_SIZE = 10;  
// 线程池维护线程所允许的空闲时间  
private final static int KEEP\_ALIVE\_TIME = 0;  
// 线程池所使用的缓冲队列大小  
private final static int WORK\_QUEUE\_SIZE = 10;  
// 消息缓冲队列  
Queue msgQueue = new LinkedList();  
// 访问消息缓存的调度线程  
final Runnable accessBufferThread = new Runnable() \{  
public void run() \{  
// 查看是否有待定请求，如果有，则创建一个新的AccessDBThread，并添加到线程池中  
if (hasMoreAcquire()) \{  
String msg = (String) msgQueue.poll();  
Runnable task = new AccessDBThread(msg);  
threadPool.execute(task);  
\}  
\}  
\};  
final RejectedExecutionHandler handler = new RejectedExecutionHandler() \{  
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) \{  
System.out.println(((AccessDBThread) r).getMsg() + "消息放入队列中重新等待执行");  
msgQueue.offer(((AccessDBThread) r).getMsg());  
\}  
\};  
// 管理数据库访问的线程池  
final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(  
CORE\_POOL\_SIZE, MAX\_POOL\_SIZE, KEEP\_ALIVE\_TIME, TimeUnit.SECONDS,  
new ArrayBlockingQueue(WORK\_QUEUE\_SIZE), this.handler);  
// 调度线程池  
final ScheduledExecutorService scheduler = Executors  
.newScheduledThreadPool(1);  
final ScheduledFuture taskHandler = scheduler.scheduleAtFixedRate(  
accessBufferThread, 0, 1, TimeUnit.SECONDS);  
  
public static ThreadPoolManager newInstance() \{  
return tpm;  
\}  
  
private ThreadPoolManager() \{  
\}  
  
private boolean hasMoreAcquire() \{  
return !msgQueue.isEmpty();  
\}  
  
public void addLogMsg(String msg) \{  
Runnable task = new AccessDBThread(msg);  
threadPool.execute(task);  
\}  
\}  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
package com.abin.message.line;  
  
public class AccessDBThread implements Runnable \{  
private String msg;  
  
public String getMsg() \{  
return msg;  
\}  
  
public void setMsg(String msg) \{  
this.msg = msg;  
\}  
  
public AccessDBThread() \{  
super();  
\}  
  
public AccessDBThread(String msg) \{  
this.msg = msg;  
\}  
  
public void run() \{  
// 向数据库中添加Msg变量值  
System.out.println("Added the message: " + msg + " into the Database");  
\}  
\}  
  
  
  
  
  
  
  
  
  
  
  
  
  
package com.abin.message.line;  
  
public class TestDriver \{  
ThreadPoolManager tpm = ThreadPoolManager.newInstance();  
  
public void sendMsg(String msg) \{  
tpm.addLogMsg(msg + "记录一条日志 ");  
\}  
  
public static void main(String\[\] args) \{  
for (int i = 0; i < 100; i++) \{  
new TestDriver().sendMsg(Integer.toString(i));  
\}  
\}  
\}