当前位置: 奥门金沙手机娱乐网址 > 服务器 > 正文

java线程池实例,java使用默认线程池踩过的坑

时间:2019-11-16 05:43来源:服务器
java使用暗中认可线程池踩过的坑(1) 场景 贰个调节器,五个调解职分,分别管理多个目录下的txt文件,某些调治职务应对有些复杂难题的时候会不停极度长的年华,以致有直接不通的恐

java使用暗中认可线程池踩过的坑(1)

场景

贰个调节器,五个调解职分,分别管理多个目录下的txt文件,某些调治职务应对有些复杂难题的时候会不停极度长的年华,以致有直接不通的恐怕。大家要求三个manager来管理这么些task,当这些task的上三回施行时间相差今后超越5个调度周期的时候,就直接停掉这么些线程,然后再重启它,保险五个指标目录下没有待管理的txt文件聚成堆。

问题

直白接收java默许的线程池调治task1和task2.由于外界txt的各种不可控原因,引致task2线程窒碍。现象正是task1和线程池调节器都例行运作着,可是task2迟迟未有动作。

当然,找到切实可行的堵截原因并开展针对消除是很要紧的。然而,这种措施很恐怕并无法一心、透彻、周到的拍卖好全部未知景况。大家须求确认保证职责线程或许调节器的强壮性!

方案陈设

线程池调节器并从未原生的对准被调节线程的作业运市场价格况进行督察处理的API。因为task2是堵塞在大家的事务逻辑里的,所以最佳的格局是写一个TaskManager,全部的天职线程在奉行职分前全体到这一个TaskManager这里来注册自个儿。这一个TaskManager就担当对于各种自身管辖范围内的task实行实时全程监督!

末端的最重要就是何等管理超越5个实践周期的task了。

方案如下:

●意气风发旦发觉那些task线程,立时暂停它,然后再度重启;

●意气风发旦发觉这么些task线程,直接将全部pool清空并终止,重新归入这五个task ——task显著的事态下】;

方案实行

暂停后重启

●Task实现类

class FileTask extends Thread { private long lastExecTime = 0; protected long interval = 10000; public long getLastExecTime() {     return lastExecTime; } public void setLastExecTime(long lastExecTime) {     this.lastExecTime = lastExecTime; } public long getInterval() {     return interval; } public void setInterval(long interval) {     this.interval = interval; }  public File[] getFiles() {     return null; } 

●Override

public void run() { while (!Thread.currentThread().isInterrupted()) { lastExecTime = System.currentTimeMillis(); System.out.println(Thread.currentThread().getName() + " is running -> " + new Date()); try { Thread.sleep(getInterval() * 6 * 1000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); e.printStackTrace();    // 当线程池shutdown之后,这里就会抛出exception了             }         }     }     } 

●TaskManager

public class TaskManager  implements Runnable { private final static Log logger = LogFactory.getLog(TaskManager .class); public Set<FileTask> runners = new CopyOnWriteArraySet<FileTask>(); ExecutorService pool = Executors.newCachedThreadPool(); public void registerCodeRunnable(FileTask process) { runners.add(process); } public TaskManager (Set<FileTask> runners) { this.runners = runners; } 

@Override

public void run() {        while (!Thread.currentThread().isInterrupted()) {            try {                long current = System.currentTimeMillis();                for (FileTask wrapper : runners) {                    if (current - wrapper.getLastExecTime() > wrapper.getInterval() * 5) {                        wrapper.interrupt();                        for (File file : wrapper.getFiles()) {                            file.delete();                        }                     wrapper.start();                      }                }            } catch (Exception e1) {                logger.error("Error happens when we trying to interrupt and restart a task ");                ExceptionCollector.registerException(e1);            }            try {                Thread.sleep(500);            } catch (InterruptedException e) {            }        }    }     

这段代码会报错 java.lang.Thread IllegalThreadStateException。为啥吧?其实那是二个很根基的主题素材,您应该不会像本人同一大体。查看Thread.start()的讲授, 有那样风姿浪漫段:

It is never legal to start a thread more than once. In particular, a thread may not be restarted once it has completed execution.

不得不承认,五个线程不可能运营五回。那么它是怎么推断的啊?

public synchronized void start() {         /**          * A zero status value corresponds to state "NEW".    0对应的是state NEW          */ 

if (threadStatus != 0) //假设不是NEW state,就平素抛出非常!


图片 1


) 场景 叁个调治器,四个调治职责,分别管理多少个目录下的txt文件,有些调治职责应对有个别复杂难题的时候会...

java中通用的线程池实例代码,java线程池实例

复制代码 代码如下:
package com.smart.frame.task.autoTask;

import java.util.Collection;
import java.util.Vector;

/**
 * 职责分发器
 */
public class TaskManage extends Thread
{
    protected Vector<Runnable> tasks = new Vector<Runnable>();
    protected boolean running = false;
    protected boolean stopped = false;
    protected boolean paused = false;
    protected boolean killed = false;
    private ThreadPool pool;

    public TaskManage(ThreadPool pool)
    {
        this.pool = pool;
    }

    public void putTask(Runnable task)
    {
        tasks.add(task);
    }

    public void putTasks(Runnable[] tasks)
    {
        for (int i = 0; i < tasks.length; i++)
            this.tasks.add(tasks[i]);
    }

    public void putTasks(Collection<Runnable> tasks)
    {
        this.tasks.addAll(tasks);
    }

    protected Runnable popTask()
    {
        if (tasks.size() > 0) return (Runnable) tasks.remove(0);
        else return null;
    }

    public boolean isRunning()
    {
        return running;
    }

    public void stopTasks()
    {
        stopped = true;
    }

    public void stopTasksSync()
    {
        stopTasks();
        while (isRunning())
        {
            try
            {
                sleep(5);
            }
            catch (InterruptedException e)
            {
                TaskException.getResultMessage(e);
            }
        }
    }

    public void pauseTasks()
    {
        paused = true;
    }

    public void pauseTasksSync()
    {
        pauseTasks();
        while (isRunning())
        {
            try
            {
                sleep(5);
            }
            catch (InterruptedException e)
            {
                TaskException.getResultMessage(e);
            }
        }
    }

    public void kill()
    {
        if (!running) interrupt();
        else killed = true;
    }

    public void killSync()
    {
        kill();
        while (isAlive())
        {
            try
            {
                sleep(5);
            }
            catch (InterruptedException e)
            {
                TaskException.getResultMessage(e);
            }
        }
    }

    public synchronized void startTasks()
    {
        running = true;
        this.notify();
    }

    public synchronized void run()
    {
        try
        {
            while (true)
            {
                if (!running || tasks.size() == 0)
                {
                    pool.notifyForIdleThread();
                    this.wait();
                }
                else
                {
                    Runnable task;
                    while ((task = popTask()) != null)
                    {
                        task.run();
                        if (stopped)
                        {
                            stopped = false;
                            if (tasks.size() > 0)
                            {
                                tasks.clear();
                                System.out.println(Thread.currentThread().getId() + ": Tasks are stopped");
                                break;
                            }
                        }
                        if (paused)
                        {
                            paused = false;
                            if (tasks.size() > 0)
                            {
                                System.out.println(Thread.currentThread().getId() + ": Tasks are paused");
                                break;
                            }
                        }
                    }
                    running = false;
                }

                if (killed)
                {
                    killed = false;
                    break;
                }
            }
        }
        catch (InterruptedException e)
        {
            TaskException.getResultMessage(e);
            return;
        }
    }
}

复制代码 代码如下:
package com.smart.frame.task.autoTask;

import java.util.Collection;
import java.util.Iterator;
import java.util.Vector;

/**
 * 线程池
 */
public class ThreadPool
{
    protected int maxPoolSize = TaskConfig.maxPoolSize;
    protected int initPoolSize = TaskConfig.initPoolSize;
    protected Vector<TaskManage> threads = new Vector<TaskManage>();
    protected boolean initialized = false;
    protected boolean hasIdleThread = false;

    public ThreadPool()
    {
        super();
    }

    public ThreadPool(int maxPoolSize, int initPoolSize)
    {
        this.maxPoolSize = maxPoolSize;
        this.initPoolSize = initPoolSize;
    }

    public void init()
    {
        initialized = true;
        for (int i = 0; i < initPoolSize; i++)
        {
            TaskManage thread = new TaskManage(this);
            thread.start();
            threads.add(thread);
        }
    }

    public void setMaxPoolSize(int maxPoolSize)
    {
        this.maxPoolSize = maxPoolSize;
        if (maxPoolSize < getPoolSize()) setPoolSize(maxPoolSize);
    }

    /**
     * 重设当前线程数 若需杀掉某线程,线程不会立刻杀掉,而会等到线程中的事
     * 务管理完了 但此方法会马上从线程池中移除该线程,不会等待事务处理甘休
     */
    public void setPoolSize(int size)
    {
        if (!initialized)
        {
            initPoolSize = size;
            return;
        }
        else if (size > getPoolSize())
        {
            for (int i = getPoolSize(); i < size && i < maxPoolSize; i++)
            {
                TaskManage thread = new TaskManage(this);
                thread.start();
                threads.add(thread);
            }
        }
        else if (size < getPoolSize())
        {
            while (getPoolSize() > size)
            {
                TaskManage th = (TaskManage) threads.remove(0);
                th.kill();
            }
        }
    }

    public int getPoolSize()
    {
        return threads.size();
    }

    protected void notifyForIdleThread()
    {
        hasIdleThread = true;
    }

    protected boolean waitForIdleThread()
    {
        hasIdleThread = false;
        while (!hasIdleThread && getPoolSize() >= maxPoolSize)
        {
            try
            {
                Thread.sleep(5);
            }
            catch (InterruptedException e)
            {
                TaskException.getResultMessage(e);
                return false;
            }
        }

        return true;
    }

    public synchronized TaskManage getIdleThread()
    {
        while (true)
        {
            for (Iterator<TaskManage> itr = threads.iterator(); itr.hasNext();)
            {
                TaskManage th = (TaskManage) itr.next();
                if (!th.isRunning()) return th;
            }

            if (getPoolSize() < maxPoolSize)
            {
                TaskManage thread = new TaskManage(this);
                thread.start();
                threads.add(thread);
                return thread;
            }

            if (waitForIdleThread() == false) return null;
        }
    }

    public void processTask(Runnable task)
    {
        TaskManage th = getIdleThread();
        if (th != null)
        {
            th.putTask(task);
            th.startTasks();
        }
    }

    public void processTasksInSingleThread(Runnable[] tasks)
    {
        TaskManage th = getIdleThread();
        if (th != null)
        {
            th.putTasks(tasks);
            th.startTasks();
        }
    }

    public void processTasksInSingleThread(Collection<Runnable> tasks)
    {
        TaskManage th = getIdleThread();
        if (th != null)
        {
            th.putTasks(tasks);
            th.startTasks();
        }
    }

}

复制代码 代码如下:
package com.smart.frame.task.autoTask;

public class TopTask implements Runnable
{

    private ThreadPool pool;

    public TopTask()
    {
        super();
    }

    public TopTask(ThreadPool pool)
    {
        super();
        this.pool = pool;
    }

    @Override
    public void run()
    {
        init();
        start();
    }

    /**
     * 开头化验证权限、参数之类
     */
    public void init()
    {

    }

    /**
     * 起头自行职务
     */
    public void start()
    {
        for (int i = 0; i < 10; i++)
        {
            pool.processTask(new BeginAuto());
        }
    }
}
/**
 * 实现类
 */
class BeginAuto implements Runnable
{
    @Override
    public void run()
    {
        System.out.println(Thread.currentThread().getId() + "..................");
    }

}

复制代码 代码如下: package com.smart.frame.task.autoTask; import java.util.Collection; import java.util.Vector;...

编辑:服务器 本文来源:java线程池实例,java使用默认线程池踩过的坑

关键词: