java自定义线程模型处理方法分享_Java教程-查字典教程网

摘要：看过我之前文章的园友可能知道我是做游戏开发，我的很多思路和出发点是按照游戏思路来处理的，所以和web的话可能会有冲突，不相符合。来说说为啥我...

看过我之前文章的园友可能知道我是做游戏开发，我的很多思路和出发点是按照游戏思路来处理的，所以和web的话可能会有冲突，不相符合。

来说说为啥我要自定义线程模型呢？

按照我做的mmorpg或者mmoarpg游戏划分，线程被划分为，主线程，全局同步线程，聊天线程，组队线程，地图线程，以及地图消息分发派送线程等；

一些列，都需要根据我的划分，以及数据流向做控制。

游戏服务器，主要要做的事情，肯定是接受玩家的命令请求 -> 相应的操作 -> 返回结果；

在服务器端所有的消息都会注册到消息管理器里，然后消息在注册的时候会指定线程模型，

如果消息需要提交到玩家所在地图线程进行处理的话注册消息的时候就要把线程模型用（地图消息分发派送线程）；

下面我们先来分析线程模型；

在看线程模型代码之前我先看看我的任务模型

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; /** * 任务模型 * * * author 失足程序员 * mail 492794628@qq.com * phone 13882122019 */ 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; /** * 定时器执行器 * * * author 失足程序员 * mail 492794628@qq.com * phone 13882122019 */ 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; /** * 线程模型 * * author 失足程序员 * mail 492794628@qq.com * phone 13882122019 */ 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("sn ") .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 + '}'; } }

我从 ThreadModel 构造函数的

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; }

可以看出，这里我运行声明一个或者多个 MyThread 线程类

为什么要这样考虑呢打个比方，如果是处理日志的写入数据这种，没有共享数据，没有线程临界区的处理流程，我可以考虑使用N个线程去处理这样的工作；不会产生脏数据；

如果是想组队请求，技能施法这种处理，我需要单队列处理，那么threadmodel里面肯定只有一个MyThread 这样不算阻塞模式串行执行（或队列执行）把共享数据和线程临界区的问题也解决了不再依赖锁；

字很丑，请见谅

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

上面图片看出，在每一个threadmodel 里面都会两个队列，一个timertaskevent，一个是taskevent，会存在一个全局的timer thread；

全局的 timer thread 的作用是用来定时去处理和发现 threadmodel里面timertaskevent需要执行了，就把他加入到taskevent队里里面；最终执行是taskevent队列

timertaskevent为什么要存储在对应的threadmodel里面呢，那是因为比如，我A线程（threadmodel实例）运行一段时间后需要关闭，释放资源了，那么我还要去其他地方查找对应的timertask并移除掉；

package net.sz.engine.thread; import java.util.HashMap; import java.util.Map; /** * * * author 失足程序员 * mail 492794628@qq.com * phone 13882122019 */ 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; /** * 线程管理器 * * * author 失足程序员 * mail 492794628@qq.com * phone 13882122019 */ 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) { @Override public 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; } /** * 向线程池注册一个线程 * * 默认分组 UnknownThreadGroup * * @param name 线程名称 * @return */ static public long addThreadModel(String name) { return addThreadModel(UnknownThreadGroup, name); } /** * 向线程池注册一个线程 * * 默认分组 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->1472120543013 run3->1472120543013 run3->1472120543017 run1->1472120543017 run2->1472120543017 run1->1472120543517 run3->1472120543517 run2->1472120543517 run1->1472120544018 run3->1472120544018 run2->1472120544018 run1->1472120544520 run3->1472120544520 run2->1472120544520 run1->1472120545021 run3->1472120545021 run2->1472120545021 run1->1472120545521 run3->1472120545521

一切正常；

这就是我的自定义线程模型；

到这里我的自定义线程模型就算介绍完成了；

那么优缺点在哪里呢？

优点是，数据流程控制很清晰，包括现在执行情况，以及线程卡死监控和任务的定时器执行；

缺点，这个自定义线程模型依然不可能解决线程数据安全和临界区问题，在适当的时候依然需要靠锁或者其他形式来解决；

不足之处希望大神们指出，我好即时纠正。