当观察者模式遇到线程

prtyaa 2023-12-30 21:27:24  53177 赞同 0 反对 0

分类：资源

在一般的开发过程中，使用线程是必不可少的，而获取线程的状态也成为了一个需求，虽然在Thread类为我们提供了线程的可获取状态，和判断alive的方法，但这些方法都是针对线程本身而言的，我们一般使用Runnable提交任务，而Runnable自身的状态是无法获取的。如果我们想要获取运行的最终结果，就不能不传入共享变量来获取，共享变量在多线程中也需要做同步处理，降低了系统性能。

当某个状态发生改变的时候，我们需要知道当前所在的状态，观察者模式就非常适合这样的场景。

定义接口：

Observable:

public interface Observable {

    //运行状态的枚举类，对应4种状态：开始，运行，结束，错误
    enum Cycle{
        STARTED,RUNNING,DONE,ERROR
    }
    //获取当前状态方法
    Cycle getCycle();
    //启动线程的方法
    void start();
    //线程打断方法
    void interrupt();

}

生命周期接口:

TaskLifecycle

public interface TaskLifecycle<T> {
    //任务启动时会触发
    void onStart(Thread thread);
    //任务运行时会触发
    void onRunning(Thread thread);
    //任务结束时会触发
    void onFinish(Thread thread,T result);
    //任务异常时会触发
    void onError(Thread thread,Exception e);

    class EmptyLifecycle<T> implements TaskLifecycle<T>{

        @Override
        public void onStart(Thread thread) {

        }

        @Override
        public void onRunning(Thread thread) {

        }

        @Override
        public void onFinish(Thread thread,T result) {

        }

        @Override
        public void onError(Thread thread,Exception e) {

        }
    }
}

自定义任务执行接口，可以有返回值

Task

@FunctionalInterface
public interface Task<T> {
    //任务进行接口
    T call();
}

具体实现

public class ObservableThread<T> extends Thread implements Observable {
    private final TaskLifecycle<T> lifecycle;
    private final Task<T> task;
    private Cycle cycle;

    //自定义Task实现，默认使用EmptyLifecycle
    public ObservableThread(Task<T> task) {
        this(new TaskLifecycle.EmptyLifecycle<T>(), task);
    }

    //自定义Task实现，同时自定义Lifecycle
    public ObservableThread(TaskLifecycle<T> lifecycle, Task<T> task) {
        super();
        if (task == null) {
            throw new IllegalArgumentException("task不能为空");
        }
        this.lifecycle = lifecycle;
        this.task = task;
    }

    @Override
    public final void run() {
        //在执行task的时候，分别触发相应的事件
        this.update(Cycle.STARTED,null,null);
        try {
            this.update(Cycle.RUNNING,null,null);
            //执行task
            T result = task.call();
            this.update(Cycle.DONE,result,null);
        }catch (Exception e){
            this.update(Cycle.ERROR,null,e);
        }
        super.run();
    }

    //生命周期变更
    public void update(Cycle cycle, T result, Exception e) {
        this.cycle = cycle;
        if (lifecycle == null) {
            return;
        }
        try {
            switch (cycle) {
                case STARTED:
                    this.lifecycle.onStart(currentThread());
                    break;
                case RUNNING:
                    this.lifecycle.onRunning(currentThread());
                    break;
                case DONE:
                    this.lifecycle.onFinish(currentThread(), result);
                    break;
                case ERROR:
                    this.lifecycle.onError(currentThread(), e);
                    break;
            }
        } catch (Exception ex) {
            if (cycle == Cycle.ERROR) {
                throw ex;
            }
        }
    }

    @Override
    public Cycle getCycle() {
        return this.cycle;
    }
}

使用与普通的Thread并没有什么区别

public class test {
    public static void main(String[] args) {
        Observable observable = new ObservableThread<>(()->{
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("完成");
            return null;
        });
        observable.start();
        observable.getCycle();
        observable.getCycle();
    }
}

ObservableThread是一个泛型类，可以定义为Void类型，表示不关心返回值。

如果您发现该资源为电子书等存在侵权的资源或对该资源描述不正确等，可点击“私信”按钮向作者进行反馈；如作者无回复可进行平台仲裁，我们会在第一时间进行处理！

评价 0 条