java自定义线程模型处理方法分享

package net.sz.engine.thread;import java.io.Serializable;import org.apache.log4j.Logger;import net.sz.engine.structs.ObjectAttribute;import net.sz.engine.structs.ObjectGlobal;/** * 任务模型 * * <br> * author 失足程序员<br> * mail 492794628@qq.com<br> * phone 13882122019<br> */public abstract class TaskEvent implements Serializable, Cloneable { private static final Logger log = Logger.getLogger(TaskEvent.class); private static final long serialVersionUID = 4196020659994845804L; //运行时数据 private transient final ObjectAttribute runOther = new ObjectAttribute; //任务创建的时间 protected long createTime; //任务的唯一id protected long taskId; //取消的任务 protected boolean cancel = false; public TaskEvent {  this.runOther.put("submitTime", System.currentTimeMillis);  createTime = System.currentTimeMillis;  cancel = false;  taskId = ObjectGlobal.getUUID; } public long getCreateTime {  return createTime; } public void setCreateTime(long createTime) {  this.createTime = createTime; } public long getSubmitTime {  return this.runOther.getlongValue("submitTime"); } public ObjectAttribute getRunOther {  return runOther; } public boolean isCancel {  return cancel; } public void setCancel(boolean cancel) {  this.cancel = cancel; } public abstract void run; @Override public Object clone throws CloneNotSupportedException {  return super.clone; //To change body of generated methods, choose Tools | Templates. }}

package net.sz.engine.thread;/** * 定时器执行器 * * <br> * author 失足程序员<br> * mail 492794628@qq.com<br> * phone 13882122019<br> */public abstract class TimerTaskEvent extends TaskEvent { private static final long serialVersionUID = -8331296295264699207L; /**  * 开始执行的时间  */ protected long startTime; /**  * 是否一开始执行一次  */ protected boolean startAction; /**  * 结束时间  */ protected long endTime; /**  * 执行次数  */ protected int actionCount; /**  * 间隔执行时间  */ protected int intervalTime; /**  *  * @param startTime 指定开始时间  * @param isStartAction 是否一开始就执行一次  * @param endTime 指定结束时间  * @param actionCount 指定执行次数  * @param intervalTime 指定间隔时间  */ public TimerTaskEvent(long startTime, boolean isStartAction, long endTime, int actionCount, int intervalTime) {  super;  this.startTime = startTime;  this.startAction = isStartAction;  this.endTime = endTime;  this.actionCount = actionCount;  this.intervalTime = intervalTime; } /**  * 指定任务的开始执行时间  *  * @param startTime 指定开始时间  * @param isStartAction 是否一开始就执行一次  * @param actionCount 指定执行次数  * @param intervalTime 指定间隔时间  */ public TimerTaskEvent(long startTime, boolean isStartAction, int actionCount, int intervalTime) {  this(startTime, isStartAction, 0, actionCount, intervalTime); } /**  * 指定结束时间已结束时间为准，执行次数不一定够  *  * @param isStartAction 是否一开始就执行一次  * @param endTime 指定结束时间  * @param actionCount 指定执行次数  * @param intervalTime 指定间隔时间  *  */ public TimerTaskEvent(boolean isStartAction, long endTime, int actionCount, int intervalTime) {  this(0, isStartAction, endTime, actionCount, intervalTime); } /**  * 指定开始时间，和结束时间  *  * @param startTime 指定开始时间  * @param endTime 指定结束时间  * @param intervalTime 指定间隔时间  */ public TimerTaskEvent(long startTime, long endTime, int intervalTime) {  this(startTime, false, endTime, -1, intervalTime); } /**  * 指定的执行次数和间隔时间  *  * @param actionCount 指定执行次数  * @param intervalTime 指定间隔时间  */ public TimerTaskEvent(int actionCount, int intervalTime) {  this(0, false, 0, actionCount, intervalTime); } /**  * 提交后指定的时间无限制执行  *  * @param intervalTime 指定间隔时间  */ public TimerTaskEvent(int intervalTime) {  this(0, false, 0, -1, intervalTime); } public long getStartTime {  return startTime; } public void setStartTime(long startTime) {  this.startTime = startTime; } public boolean isStartAction {  return startAction; } public void setStartAction(boolean startAction) {  this.startAction = startAction; } public long getEndTime {  return endTime; } public void setEndTime(long endTime) {  this.endTime = endTime; } public int getActionCount {  return actionCount; } public void setActionCount(int actionCount) {  this.actionCount = actionCount; } public int getIntervalTime {  return intervalTime; } public void setIntervalTime(int intervalTime) {  this.intervalTime = intervalTime; }}

package net.sz.engine.thread;import java.util.ArrayList;import java.util.List;import java.util.concurrent.ConcurrentLinkedQueue;import net.sz.engine.structs.ObjectGlobal;import net.sz.engine.utils.MailUtil;import net.sz.engine.utils.StringUtil;import org.apache.log4j.Logger;import org.jboss.jandex.Main;/** * 线程模型 * <br> * author 失足程序员<br> * mail 492794628@qq.com<br> * phone 13882122019<br> */public class ThreadModel implements Runnable { private static final Logger log = Logger.getLogger(ThreadModel.class); private static long threadID = 0; protected static final Object SYN_OBJECT = new Object; protected long tid; protected String name; protected long lastSendMail = 0; protected final ArrayList<MyThread> threads = new ArrayList<>; /**  * 任务列表 线程安全的任务列表  */ //protected final List<TaskModel> taskQueue = new ArrayList<>; protected final ConcurrentLinkedQueue<TaskEvent> taskQueue = new ConcurrentLinkedQueue<>; /**  *  */ protected final List<TimerTaskEvent> timerQueue = new ArrayList<>; // false标识删除线程 protected volatile boolean runing = true; public ThreadModel(ThreadGroup group) {  this(group, "无名", 1); } public ThreadModel(String name) {  this(ThreadPool.UnknownThreadGroup, name, 1); } public ThreadModel(ThreadGroup group, String name, int threadCount) {  this(group, name, threadCount, null); } public ThreadModel(ThreadGroup group, String name, int threadCount, Runnable runnable) {  synchronized (SYN_OBJECT) {threadID++;tid = threadID;  }  for (int i = 1; i <= threadCount; i++) { MyThread thread;if (runnable == null) {thread = new MyThread(tid, group, this, name + "-" + tid + "-" + i);} else {thread = new MyThread(tid, group, runnable, name + "-" + tid + "-" + i); } thread.start; threads.add(thread);  }  this.name = name; } /**  * 线程名字  *  * @return  */ public String getName {  return name; } /**  * 获取线程的自定义id  *  * @return  */ public long getId {  return this.tid; } /**  * 增加新的任务 每增加一个新任务，都要唤醒任务队列  *  * @param runnable  */ public void addTask(TaskEvent runnable) {  taskQueue.add(runnable);  synchronized (taskQueue) {/* 唤醒队列, 开始执行 */ taskQueue.notifyAll;  } } /**  * 向线程添加定时器任务  *  * @param runnable  */ public void addTimer(TimerTaskEvent runnable) {  synchronized (timerQueue) {if (runing) {//一开始执行一次if (runnable.startAction) { addTask(runnable); } timerQueue.add(runnable);} else {log.error("线程已经停止"); }  } } // <editor-fold defaultstate="collapsed" desc="定时器线程执行器 public void timerRun"> /**  * 定时器线程执行器  */ public void timerRun {  ArrayList<TimerTaskEvent> taskModels;  synchronized (timerQueue) {// 队列不为空的情况下 取出队列定时器任务taskModels = new ArrayList<>(timerQueue);  }  if (!taskModels.isEmpty) {for (TimerTaskEvent timerEvent : taskModels) {int execCount = timerEvent.getRunOther.getintValue("Execcount");long lastTime = timerEvent.getRunOther.getlongValue("LastExecTime");long nowTime = System.currentTimeMillis;if (lastTime == 0) {timerEvent.getRunOther.put("LastExecTime", nowTime);} else if (timerEvent.isCancel) {//如果任务已经取消synchronized (timerQueue) { timerQueue.remove(timerEvent); }log.debug("清理定时器任务：" + timerEvent.getClass.getName);} else if (nowTime > timerEvent.getStartTime // 是否满足开始时间&& (nowTime - timerEvent.getSubmitTime > timerEvent.getIntervalTime)// 提交以后是否满足了间隔时间&& (timerEvent.getEndTime <= 0 || nowTime < timerEvent.getEndTime) // 判断结束时间&& (nowTime - lastTime >= timerEvent.getIntervalTime)) // 判断上次执行到目前是否满足间隔时间 {// 提交执行定时器最先执行this.addTask(timerEvent);// 记录execCount++;timerEvent.getRunOther.put("Execcount", execCount);timerEvent.getRunOther.put("LastExecTime", nowTime);nowTime = System.currentTimeMillis;// 判断删除条件if ((timerEvent.getEndTime > 0 && nowTime < timerEvent.getEndTime)|| (timerEvent.getActionCount > 0 && timerEvent.getActionCount <= execCount)) {synchronized (timerQueue) { timerQueue.remove(timerEvent); }log.debug("清理定时器任务：" + timerEvent.getClass.getName); } } }  } } // </editor-fold> // <editor-fold defaultstate="collapsed" desc="查看线程堆栈 public void showStackTrace"> /**  *  * 查看线程堆栈  */ public void showStackTrace {  StringBuilder buf = new StringBuilder;  for (MyThread currentThread : threads) {long procc = System.currentTimeMillis - currentThread.getLastExecuteTime;if (procc > 5 * 1000 && procc < 864000000L) {//小于10天//因为多线程操作时间可能不准确buf.append("线程[") .append(currentThread.getName).append("]可能已卡死 -> ").append(procc / 1000f).append("s/n ").append("执行任务：") .append(currentThread.getLastCommand.getClass.getName);try {StackTraceElement elements = currentThread.getStackTrace;for (int i = 0; i < elements.length; i++) {buf.append("/n ") .append(elements[i].getClassName).append(".") .append(elements[i].getMethodName).append("(").append(elements[i].getFileName).append(";").append(elements[i].getLineNumber).append(")"); }} catch (Exception e) { buf.append(e); }buf.append("/n++++++++++++++++++++++++++++++++++"); }  }  String toString = buf.toString;  if (!StringUtil.isNullOrEmpty(toString)) { log.error(toString);if (System.currentTimeMillis - lastSendMail > 5 * 60 * 1000) {lastSendMail = System.currentTimeMillis;MailUtil.sendMail("线程执行已卡死 -> 游戏id-" + ObjectGlobal.GameID + " 平台-" + ObjectGlobal.Platform + " 服务器id-" + ObjectGlobal.ServerID, toString); }  } } // </editor-fold> @Override public void run {  MyThread currentThread = (MyThread) Thread.currentThread;  while (runing) {while (taskQueue.isEmpty && runing) {try {/* 任务队列为空，则等待有新任务加入从而被唤醒 */synchronized (taskQueue) {taskQueue.wait(500); }} catch (InterruptedException ie) { log.error(ie); } }/* 取出任务执行 */if (runing) {currentThread.lastCommand = null;currentThread.lastCommand = taskQueue.poll; }if (currentThread.lastCommand != null) {if (currentThread.lastCommand.isCancel) {//如果任务已经取消continue; }/* 执行任务 */// r.setSubmitTimeL;currentThread.lastExecuteTime = System.currentTimeMillis;try { currentThread.lastCommand.run;} catch (Exception e) {log.error("工人<“" + currentThread.getName + "”> 执行任务<" + currentThread.lastCommand.getClass.getName + "> 遇到错误: ", e); }long timeL1 = System.currentTimeMillis - currentThread.lastExecuteTime;if (timeL1 <= 20) {} else if (timeL1 <= 100L) {log.info("工人<“" + currentThread.getName + "”> 完成了任务：" + currentThread.lastCommand.toString + " 执行耗时：" + timeL1);} else if (timeL1 <= 200L) {log.info("工人<“" + currentThread.getName + "”> 长时间执行 完成任务：" + currentThread.lastCommand.toString + " “考虑”任务脚本逻辑 耗时：" + timeL1);} else {log.info("工人<“" + currentThread.getName + "”> 超长时间执行完成 任务：" + currentThread.lastCommand.toString + " “考虑是否应该删除”任务脚本 耗时：" + timeL1); }currentThread.lastExecuteTime = 0; }  }  log.error("线程结束, 工人<“" + Thread.currentThread.getName + "”>退出"); } /**  * 自定义线程  */ public class MyThread extends Thread {  /**   *   * @param tid 自定义线程id   * @param group 分组   * @param run 执行方法   * @param name 线程名称*/  public MyThread(long tid, ThreadGroup group, Runnable run, String name) {super(group, run, name);this._id = tid;  }  //线程的自定义id  public long _id;  //正在执行的任务  public volatile TaskEvent lastCommand;  //开始执行任务的时间  public volatile long lastExecuteTime = 0;  public TaskEvent getLastCommand {return lastCommand;  }  public long getLastExecuteTime {return lastExecuteTime;  }  /**   * 返回线程自定义id   *   * @return*/  @Override  public long getId {return _id;  } } /**  * 停止线程，设置线程的停止状态，并不会马上终止线程  */ public void stop {  this.runing = false; } public boolean isRuning {  return runing; } @Override public String toString {  return "Thread{" + "tid=" + tid + ",Name=" + this.getName + '}'; }}

 public ThreadModel(ThreadGroup group, String name, int threadCount, Runnable runnable) {  synchronized (SYN_OBJECT) {threadID++;tid = threadID;  }  for (int i = 1; i <= threadCount; i++) { MyThread thread;if (runnable == null) {thread = new MyThread(tid, group, this, name + "-" + tid + "-" + i);} else {thread = new MyThread(tid, group, runnable, name + "-" + tid + "-" + i); } thread.start; threads.add(thread);  }  this.name = name; }

package net.sz.engine.thread;import java.util.HashMap;import java.util.Map;/** * * <br> * author 失足程序员<br> * mail 492794628@qq.com<br> * phone 13882122019<br> */class TimerThread extends Thread { private static final Object SYN_OBJECT = new Object; public TimerThread {  super(ThreadPool.GloblThreadGroup, "Global Timer Thread"); } @Override public void run {  while (true) {synchronized (SYN_OBJECT) {try {SYN_OBJECT.wait(2);} catch (InterruptedException ex) { } }HashMap<Long, ThreadModel> hashMap = new HashMap<>(ThreadPool.getThreadMap);for (Map.Entry<Long, ThreadModel> entrySet : hashMap.entrySet) {Long key = entrySet.getKey;ThreadModel value = entrySet.getValue; value.timerRun; }  } }}

package net.sz.engine.thread;import java.util.HashMap;import java.util.concurrent.ConcurrentHashMap;import net.sz.engine.script.manager.ScriptManager;import net.sz.engine.timer.GlobTimerEvent;import net.sz.engine.timer.PrintlnServerMemoryTimerEvent;import org.apache.log4j.Logger;/** * 线程管理器 * * <br> * author 失足程序员<br> * mail 492794628@qq.com<br> * phone 13882122019<br> */public class ThreadPool { static private final Logger log = Logger.getLogger(ThreadPool.class); static public final long GloblThread; static private final TimerThread GloblTimerThread; static final long CheckThreadTimerThreadModel; static public final ThreadGroup GloblThreadGroup = new ThreadGroup("Global ThreadGroup"); static public final ThreadGroup UnknownThreadGroup = new ThreadGroup(GloblThreadGroup, "Unknown ThreadGroup"); static private final ConcurrentHashMap<Long, ThreadModel> threadMap = new ConcurrentHashMap<>; public static void main(String[] args) {  ThreadPool.addTimerTask(GloblThread, new TimerTaskEvent(1000) { @Overridepublic void run {log.error("ssssss"); }  }); } static {  //创建全局线程  GloblThread = addThreadModel(GloblThreadGroup, "GloblThread");  //执行指定任务定时触发脚步  addTimerTask(GloblThread, new GlobTimerEvent(ScriptManager.getInstance.getBaseScriptEntry));  //查询服务器消耗定时模型  addTimerTask(GloblThread, new PrintlnServerMemoryTimerEvent);  //创建定时器线程  GloblTimerThread = new TimerThread;  GloblTimerThread.start;  //检查线程卡死情况  CheckThreadTimerThreadModel = addThreadModel(GloblThreadGroup, "Check Thread Timer Event");  addTimerTask(CheckThreadTimerThreadModel, new CheckThreadTimerEvent); } /**  * 删除指定id线程模型的时候回设置状态为停止状态  *  * @param tid  * @return  */ static public ThreadModel remove(long tid) {  ThreadModel remove = threadMap.remove(tid);  if (remove != null) { remove.stop;  }  return remove; } /**  * 获取线程池中所有线程  *  * @return  */ static public ConcurrentHashMap<Long, ThreadModel> getThreadMap {  return threadMap; } /**  * 获取线程池的一个线程  *  * @param threadId  * @return  */ static public ThreadModel getThreadModel(long threadId) {  ThreadModel get = threadMap.get(threadId);  if (get == null) {log.error("无法找到线程模型：" + threadId, new Exception("无法找到线程模型：" + threadId));  }  return get; } /**  * 向线程池注册一个线程  * <br>  * 默认分组 UnknownThreadGroup  *  * @param name 线程名称  * @return  */ static public long addThreadModel(String name) {  return addThreadModel(UnknownThreadGroup, name); } /**  * 向线程池注册一个线程  * <br>  * 默认分组 UnknownThreadGroup  *  * @param name 线程名称  * @param threadcount 线程量  * @return  */ static public long addThreadModel(String name, int threadcount) {  return addThreadModel(UnknownThreadGroup, name, threadcount); } /**  * 向线程池注册一个线程  *  * @param group 线程分组信息  * @param name 线程名称  * @return  */ static public long addThreadModel(ThreadGroup group, String name) {  return addThreadModel(group, name, 1); } /**  * 向线程池注册一个线程  *  * @param group 线程分组信息  * @param name 线程名称  * @param threadcount 线程量  * @return  */ static public long addThreadModel(ThreadGroup group, String name, int threadcount) {  return addThreadModel(group, name, null, threadcount); } /**  * 向线程池注册一个线程  *  * @param group 线程分组信息  * @param name 线程名称  * @param runnable  * @param threadcount 线程量  * @return  */ static public long addThreadModel(ThreadGroup group, String name, Runnable runnable, int threadcount) {  ThreadModel threadModel = new ThreadModel(group, name, threadcount, runnable);  return addThreadModel(threadModel); } /**  * 向线程池注册一个线程  *  * @param threadModel  */ static public long addThreadModel(ThreadModel threadModel) {  threadMap.put(threadModel.getId, threadModel);  return threadModel.getId; } /**  * 添加任务  *  * @param threadId  * @param task  * @return  */ static public boolean addTask(long threadId, TaskEvent task) {  ThreadModel threadModel = getThreadModel(threadId);  if (threadModel != null) { threadModel.addTask(task);return true;  }  return false; } /**  * 添加定时器任务  *  * @param threadId  * @param task  * @return  */ static public boolean addTimerTask(long threadId, TimerTaskEvent task) {  ThreadModel threadModel = getThreadModel(threadId);  if (threadModel != null) { threadModel.addTimer(task);return true;  }  return false; } /**  * 添加任务,添加任务到当前线程  *  * @param task  * @return  */ static public boolean addCurrentThreadTask(TaskEvent task) {  Thread currentThread = Thread.currentThread;  if (currentThread instanceof ThreadModel.MyThread) {long threadId = currentThread.getId;ThreadModel threadModel = getThreadModel(threadId);if (threadModel != null) { threadModel.addTask(task);return true; }  }  return false; } /**  * 添加定时器任务，添加任务到当前线程  *  * @param task  * @return  */ static public boolean addCurrentThreadTimerTask(TimerTaskEvent task) {  Thread currentThread = Thread.currentThread;  if (currentThread instanceof ThreadModel.MyThread) {long threadId = currentThread.getId;ThreadModel threadModel = getThreadModel(threadId);if (threadModel != null) { threadModel.addTimer(task);return true; }  }  return false; }}

public static void main(String[] args) throws InterruptedException {  //线程并行情况，有多个线程执行多个任务/函数  new Thread(new Run1).start;  new Thread(new Run2).start; } //任务1 static class Run1 implements Runnable {  @Override  public void run {//执行任务1 run1;//执行任务3 run3;  } } //任务2 static class Run2 implements Runnable {  @Override  public void run {//执行任务3 run3;//执行任务1 run1;//执行任务2 run2;  } } //任务1 public static void run1 {  System.out.println("run1->" + System.currentTimeMillis); } //任务2 public static void run2 {  System.out.println("run2->" + System.currentTimeMillis); } //任务3 public static void run3 {  System.out.println("run3->" + System.currentTimeMillis); }

 public static void main(String[] args) throws InterruptedException {  //线程并行情况，有多个线程执行多个任务/函数  long test1 = ThreadPool.addThreadModel("测试线程-1");  long test2 = ThreadPool.addThreadModel("测试线程-2");  //添加任务  ThreadPool.addTask(test1, new Run1);  ThreadPool.addTask(test2, new Run2);  //添加定时器任务  ThreadPool.addTimerTask(test1, new TimerRun1);  ThreadPool.addTimerTask(test2, new TimerRun2); } //任务1 static class Run1 extends TaskEvent {  @Override  public void run {//执行任务1 run1;//执行任务3 run3;  } } //任务1 static class TimerRun1 extends TimerTaskEvent {  public TimerRun1 {super(500);//500毫秒无限制执行  }  @Override  public void run {//执行任务1 run1;//执行任务3 run3;  } } //任务2 static class Run2 extends TaskEvent {  @Override  public void run {//执行任务3 run3;//执行任务1 run1;//执行任务2 run2;  } } //任务2 static class TimerRun2 extends TimerTaskEvent {  public TimerRun2 {super(500);//500毫秒无限制执行  }  @Override  public void run {//执行任务3 run3;//执行任务1 run1;//执行任务2 run2;  } } //任务1 public static void run1 {  System.out.println("run1->" + System.currentTimeMillis); } //任务2 public static void run2 {  System.out.println("run2->" + System.currentTimeMillis); } //任务3 public static void run3 {  System.out.println("run3->" + System.currentTimeMillis); }

run1->1472120543013run3->1472120543013run3->1472120543017run1->1472120543017run2->1472120543017run1->1472120543517run3->1472120543517run2->1472120543517run1->1472120544018run3->1472120544018run2->1472120544018run1->1472120544520run3->1472120544520run2->1472120544520run1->1472120545021run3->1472120545021run2->1472120545021run1->1472120545521run3->1472120545521