浅析Java之Synchronized

2022-08-30

为什么线程中断不了？(已收录)

2022-08-28

前言

想必大家都知道 interrupt()函数，从这个单词上【interrupt 英[ˌɪntəˈrʌpt]，美[ˌɪntəˈrʌpt]，desc：打断; 打扰; 插嘴; 打岔; 使暂停; 使中断; 阻断，遮挡】，看起来是中断的含义，那么它真的能让运行到一半的线程中止掉吗？

Interrupt 函数

https://www.matools.com/file/manual/jdk_api_1.8_google/

据上图根据官方解释，在调用 interrupt 函数以后，收到 InterruptedException 异常的前提是该线程“阻塞的调用 wait ， sleep ，或 join 的方法”，那么假设某线程不在其列出的阻塞调用中，是否能使用此方法中止并抛出 InterruptedException 呢？实践出真知，直接操作一把，看能否中断便一目了然。

/**
 * @author Duansg
 * @date 2022-08-28 10:19 下午
 */
@SuppressWarnings("all")
public class InterruptMain {

    public static void main(String[] args) {
        final Thread thread = new Thread(() -> {
            for (; ; ) {
                System.out.println("exit");
            }
        });
        thread.start();
        thread.interrupt();
        System.out.println("main stop");
    }
}

根据伪代码执行并不能中止，也不会抛出 InterruptedException，如按官方文档所描述，增加调用阻塞方法即可得到 InterruptedException，【需要注意的是在阻塞的情况中断，不仅会得到中断异常，还会恢复中断标志位】但是子线程还是会继续输出“exit”，那么如何中断？

import java.util.concurrent.TimeUnit;

/**
 * @author Duansg
 * @date 2022-08-28 10:19 下午
 */
@SuppressWarnings("all")
public class InterruptMain {

    public static void main(String[] args) {
        final Thread thread = new Thread(() -> {
            for (; ; ) {
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    System.out.println("InterruptedException exit");
                    break;
                }
            }
        });
        thread.start();
        thread.interrupt();
        System.out.println("main stop");
    }
}

由此上代码，捕获了 InterruptedException，然后手动停止了当前线程，到此不难理解 interrupt 函数向某个线程发送中断请求，这个请求操作是将被请求线程的中断标志位设置为 true，重要的是它仅仅只是发送中断请求，至于这个线程能否被中断，这取决于被请求线程是否检查了中断标志位[可以使用 isInterrupted()]或者对中断异常做出对应的响应。

为什么是轻量级阻塞？

java 线程的状态

// java.lang.Thread.State
public enum State {
        /**
         * Thread state for a thread which has not yet started.
         */
        NEW,

        /**
         * Thread state for a runnable thread.  A thread in the runnable
         * state is executing in the Java virtual machine but it may
         * be waiting for other resources from the operating system
         * such as processor.
         */
        RUNNABLE,

        /**
         * Thread state for a thread blocked waiting for a monitor lock.
         * A thread in the blocked state is waiting for a monitor lock
         * to enter a synchronized block/method or
         * reenter a synchronized block/method after calling
         * {@link Object#wait() Object.wait}.
         */
        BLOCKED,

        /**
         * Thread state for a waiting thread.
         * A thread is in the waiting state due to calling one of the
         * following methods:
         * <ul>
         *   <li>{@link Object#wait() Object.wait} with no timeout</li>
         *   <li>{@link #join() Thread.join} with no timeout</li>
         *   <li>{@link LockSupport#park() LockSupport.park}</li>
         * </ul>
         *
         * <p>A thread in the waiting state is waiting for another thread to
         * perform a particular action.
         *
         * For example, a thread that has called <tt>Object.wait()</tt>
         * on an object is waiting for another thread to call
         * <tt>Object.notify()</tt> or <tt>Object.notifyAll()</tt> on
         * that object. A thread that has called <tt>Thread.join()</tt>
         * is waiting for a specified thread to terminate.
         */
        WAITING,

        /**
         * Thread state for a waiting thread with a specified waiting time.
         * A thread is in the timed waiting state due to calling one of
         * the following methods with a specified positive waiting time:
         * <ul>
         *   <li>{@link #sleep Thread.sleep}</li>
         *   <li>{@link Object#wait(long) Object.wait} with timeout</li>
         *   <li>{@link #join(long) Thread.join} with timeout</li>
         *   <li>{@link LockSupport#parkNanos LockSupport.parkNanos}</li>
         *   <li>{@link LockSupport#parkUntil LockSupport.parkUntil}</li>
         * </ul>
         */
        TIMED_WAITING,

        /**
         * Thread state for a terminated thread.
         * The thread has completed execution.
         */
        TERMINATED;
    }

java 线程状态的流转过程

此上我们不难看出如果调用了阻塞函数，线程都会进入 WAITTING/TIMED_WAITTING 状态，也只有在这个状态中的线程，我们才可以进行响应中断，所以如果你看到这里，你可以将其 interrupt 函数理解为“唤醒一个轻量阻塞的线程”。

为什么不是重量级阻塞？

为什么处于 BLOCKED 状态的线程不能响应中断？大家可以去看看 Thread 提供的阻塞函数中，InterruptedException 作为一个检查异常而存在，也就是相当于我们在方法中 throw new InterruptedException();而 synchronized 作为一个关键字，是无法抛出检查异常的。而一个线程排查死锁问题，其不会一致处在 BLOCKED 状态中的，只有等 BLOCKED 恢复到运行状态，通过响应中断进行处理。

如何检查中断？

除了可以捕获 InterruptedException 进行响应中断，我们还可以根据提供的两个方法进行检查中断标志位，如下。

为什么 interrupted()需要清除中断标志

在大部分场景下，我们一旦检测到中断，会进行响应处理，也就是此次的中断我已经响应并处理了，如果不清除，那么假设线程还在执行，程序上不仅要处理循环检查等问题，还无法判断中断次数，清除本次中断标志，有助于感知下次中断。

是响应中断异常还是抛出异常？

这个问题取决于你的逻辑处理，甚至你可以输出摘要日志进行人工补偿，但其处理方式，笔者认为跟其他处理异常方式无差别，当你明确知道如何处理这个异常的时候就可以捕获，如果当前处理不了就抛出或者忽略。

在谈中断的意义

中断其目的是为了优雅的停止某个线程，能让线程感知自身已经被列为中断线程，然后进行一些必要逻辑的处理。假设没有响应中断，直接调用 Thread.stop() 方法【已过时】，线程直接被杀掉，那么这个线程可能拥有的资源因为非正常的关闭导致一系列的问题发生。

本文回顾

interrupt()方法：中断(“唤醒”)轻量级阻塞线程，仅仅发送请求给线程设置一个中断标志，但是线程依旧会执行。
interrupted()方法：Thread 类的静态方法。检查当前线程的中断标志，返回一个 boolean 并清除中断状态，其连续两次调用的返回结果不一样，因为第二次调用的时候线程的中断状态已经被清除，会返回一个 false。
isInterrupted()方法：测试线程是否被中断，不会清除中断状态。

展开全文 >>

如何优雅的关闭线程池？(已收录)

2022-08-27

前言

使用自创建的线程跟使用线程池有啥区别?提到线程池是不是脑海中闪现了创建线程池的那几个核心参数、工作流程、线程池的复用、拒绝机制、缓冲机制等，这些理论知识点想必也牢记许久了。虽然线程池支持在虚拟机进程接受到退出命令后可以进行 shutdown。那么 shutdown 跟线程中断又有什么区别？在运行中的线程能否直接 kill 掉？我们能否监听关闭事件进行补偿？

为什么线程中断不了？

回答这个问题之前需要先思考，中断会有什么影响？如果直接 kill 掉，那么这个线程使用的所有资源能被正常释放吗？我们虽然 interrupt()方法是中断线程的没错,但是它也仅仅是将线程的中断位设置为 true,它不会停止线程,而是需要用户自己去监视线程的状态为并做对应的处理。可能你会说 stop()方法,这个方法已经被废弃掉了,这种显示的停止的方法带来的问题会更多,你可以想象一下会有哪些影响？言归正传,如果线程没有对中断进行处理,仅仅是调用 interrupt()是不会停止的,如下。

public static void main(String[] args) {
    Thread thread = new Thread(() -> {
        while (true){
            if(Thread.currentThread().isInterrupted()){
                System.out.println("Error");
            }else {
                System.out.println("Success");
            }
        }
    });
    thread.start();
    //ps:此代码没有做响应中断处理。
    thread.interrupt();
}

如何解决任务丢失？

前面说了这么多,现在看来,抛开直接 KILL 掉程序不说,首先我们要保证程序在正常的重启期间,任务是不能丢失的,你可能先想到是实现 Hook 方法,在程序关闭的时候触发收尾工作,来保证线程池的正常关闭。老师傅说可以交给 Spring 管理即可,思考了一下也是,我们只需要实现 destroy 方法,然后 shutdown 即可。既然想到了 Spring,那 Spring 的线程池是否处理了 shutdown 方法,那我们来探究一下,日常使用 Spring 线程池的时候最常见的一行配置你肯定见过,如下。

<bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
  <!-- 线程池维护线程的最少数量 -->
  <property name="corePoolSize" value="5" />
  <!-- 允许的空闲时间 -->
  <property name="keepAliveSeconds" value="200" />
  <!-- 线程池维护线程的最大数量 -->
  <property name="maxPoolSize" value="10" />
  <!-- 缓存队列 -->
  <property name="queueCapacity" value="20" />
  <!-- 对拒绝task的处理策略 -->
  <property name="rejectedExecutionHandler">
    <bean class="java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy" />
  </property>
</bean>

ThreadPoolTaskExecutor

public class ThreadPoolTaskExecutor extends ExecutorConfigurationSupport implements AsyncListenableTaskExecutor, SchedulingTaskExecutor {

    public void setCorePoolSize(int corePoolSize) {
        //...
    }
    public int getCorePoolSize() {
        //...
    }
    public void setMaxPoolSize(int maxPoolSize) {
        //...
    }
    //...
}

InitializingBean 跟 DisposableBean 这不经常见么,是的没错,凡是继承 InitializingBean 跟 DisposableBean 的 Bean，会在 bean 的初始化跟销毁的时候调用对应的 afterPropertiesSet()跟 destory()方法。具体直接上一张图搞定。

ExecutorConfigurationSupport

初始化方法暂且抛开不说,因为我们这里探究的是销毁方法。直接上代码。

public void destroy() {
        this.shutdown();
  }

  public void shutdown() {
        if (this.logger.isInfoEnabled()) {
            this.logger.info("Shutting down ExecutorService" + (this.beanName != null ? " '" + this.beanName + "'" : ""));
        }
        if (this.executor != null) {
           //waitForTasksToCompleteOnShutdown为被调用时是否等待当前任务完成
            if (this.waitForTasksToCompleteOnShutdown) {
            //如果需要等待任务执行完,就调用shutdown方法
                this.executor.shutdown();
            } else {
            //如果不需求,就强制关闭,并会返回那些未执行的任务
                Iterator var1 = this.executor.shutdownNow().iterator();
                while(var1.hasNext()) {
                    Runnable remainingTask = (Runnable)var1.next();
                    //取消所有剩下需要执行的线程
                    this.cancelRemainingTask(remainingTask);
                }
            }
      //如果设置了等待时长,当前线程阻塞直到如下3种情况
       //1:等所有已提交的任务（包括正在运行的和队列中等待的）执行完毕
       //2:超过设置时长
       //3:线程被中断,响应InterruptedException
            this.awaitTerminationIfNecessary(this.executor);
        }
    }

由此上我们可以看到,使用 Spring 线程池可以最大程度上解决任务丢失的问题,当然我们可以通过实现 DisposableBean 接口来自定义销毁的操作。那么还有更优雅的处理方式吗？

Guava

我相信很多童鞋都用过 Guava,Guava 中包含了许多并发类,同时也包含了几个方便的线程池相关的 ExecuorService 的实现,但需要注意的是这些实现类都无法通过直接创建或者子类化来创建实例,但是我们可以通过 MoreExecuors 来创建它们,MoreExecuors 中的方法如下

依赖

<dependency>
  <groupId>com.google.guava</groupId>
  <artifactId>guava</artifactId>
  <version>20.0</version>
</dependency>

getExitingExecutorService

//获得一个跟随JVM关闭而关闭的线程池
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10, 5L, TimeUnit.SECONDS,
                                                               new LinkedBlockingDeque<>(20), new ThreadPoolExecutor.CallerRunsPolicy());
ExecutorService executorService = MoreExecutors.getExitingExecutorService(threadPoolExecutor);

源码分析

//关键源码
final ExecutorService getExitingExecutorService(ThreadPoolExecutor executor) {
            //120S为关闭等待时长
            return this.getExitingExecutorService(executor, 120L, TimeUnit.SECONDS);
}
final ExecutorService getExitingExecutorService(ThreadPoolExecutor executor, long terminationTimeout, TimeUnit timeUnit) {
            //改变ThreadFactory,为它设置守护线程,
            MoreExecutors.useDaemonThreadFactory(executor);
            //封装传入的线程池
            ExecutorService service = Executors.unconfigurableExecutorService(executor);
            //设置钩子线程,并设置等待时长
            this.addDelayedShutdownHook(service, terminationTimeout, timeUnit);
            return service;
}

final void addDelayedShutdownHook(final ExecutorService service, final long terminationTimeout, final TimeUnit timeUnit) {
     Preconditions.checkNotNull(service);
     Preconditions.checkNotNull(timeUnit);
     this.addShutdownHook(MoreExecutors.newThread("DelayedShutdownHook-for-" + service, new Runnable() {
         public void run() {
              try {
                 //关闭线程池
                 service.shutdown();
                 //这个需要等到提交的任务全部执行完,这个上文跟Spring的阻塞等待关闭是一个道理
                 service.awaitTermination(terminationTimeout, timeUnit);
               } catch (InterruptedException var2) {
               }
         }
     }));
}

addDelayedShutdownHook

1 2	//方法使用 MoreExecutors.addDelayedShutdownHook(threadPoolExecutor,120L,TimeUnit.SECONDS);

源码分析

//关键源码,此处调用的addDelayedShutdownHook跟上面的一样,只不过这里没有改变ThreadFactory,只设置了钩子线程
public static void addDelayedShutdownHook(ExecutorService service, long terminationTimeout, TimeUnit timeUnit) {
        (new MoreExecutors.Application()).addDelayedShutdownHook(service, terminationTimeout, timeUnit);
}

shutdownAndAwaitTermination

1 2	//使用方法 MoreExecutors.shutdownAndAwaitTermination(threadPoolExecutor,10L,TimeUnit.SECONDS);

源码分析


//关键源码
public static boolean shutdownAndAwaitTermination(ExecutorService service, long timeout, TimeUnit unit) {
        long halfTimeoutNanos = unit.toNanos(timeout) / 2L;
        //预关闭
        service.shutdown();
        try {
            //判断是否有提交的任务没有执行完。
            if (!service.awaitTermination(halfTimeoutNanos, TimeUnit.NANOSECONDS)) {
                //如果没有执行完,继续尝试强行关闭
                service.shutdownNow();
                //然后进行阻塞等待关闭
                service.awaitTermination(halfTimeoutNanos, TimeUnit.NANOSECONDS);
            }
        } catch (InterruptedException var7) {
            //如果在关闭过程中如果发生中断,先设置中断恢复。
            Thread.currentThread().interrupt();
            //直接强行关闭
            service.shutdownNow();
        }
        //返回状态
        return service.isTerminated();
}

Example-Demo

//1
@Configuration
public class ThreadPoolConfiguration {

    @Bean(name = "executorServiceExiting")
    public ExecutorService executorService(){
        ThreadPoolExecutor threadPoolExecutor =
          new ThreadPoolExecutor(5, 10, 5L, TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(20), new ThreadPoolExecutor.CallerRunsPolicy());
        return MoreExecutors.getExitingExecutorService(threadPoolExecutor);
    }

    @Bean(name = "executorServiceShutdownHook")
    public ExecutorService executorServiceShutdownHook(){
        ThreadPoolExecutor threadPoolExecutor =
          new ThreadPoolExecutor(5, 10, 5L, TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(20), new ThreadPoolExecutor.CallerRunsPolicy());
        MoreExecutors.addDelayedShutdownHook(threadPoolExecutor,120L,TimeUnit.SECONDS);
        return threadPoolExecutor;
    }
}

//2
@Configuration
public class ThreadPoolConfiguration {


    @Bean
    @Order(2)
    public ExecutorService executorService(){
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10, 5L, TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(20), new ThreadPoolExecutor.CallerRunsPolicy());
        ThreadPoolRegisterCenter.register(threadPoolExecutor);
        return threadPoolExecutor;
    }
}
//-------------
/**
 * @author Duansg
 * @desc 线程池的注册中心
 * @date 2020-04-01 00:23:12
 */
@Configuration
@Order(1)
public class ThreadPoolRegisterCenter implements ApplicationRunner, DisposableBean {
    /**
     * @desc 线程池容器
     */
    private static final Set<ExecutorService> THREAD_POOL_CONTEXT  = Sets.newConcurrentHashSet();

    /**
     * @desc 添加线程池到容器中
     * @param executorService
     */
    public static void register(ExecutorService executorService) {
        THREAD_POOL_CONTEXT.add(executorService);
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        //设置钩子方法,并使用shutdownAndAwaitTermination方法进行关闭处理
        Runtime.getRuntime().addShutdownHook(new Thread(() -> THREAD_POOL_CONTEXT.forEach(executorService -> {
            if (!ObjectUtils.isEmpty(executorService)){
                shutdownAndAwaitTermination(executorService);
            }
        })));
    }

    @Override
    public void destroy() throws Exception {
        //防止调用tomcat reload方法,导致的无法感知到关闭事件
        THREAD_POOL_CONTEXT.forEach(executorService -> {
            if (shutdownAndAwaitTermination(executorService)){
                THREAD_POOL_CONTEXT.remove(executorService);
            }
        });
    }

    /**
     * @desc 关闭方法
     * @param executorService
     */
    private boolean shutdownAndAwaitTermination(ExecutorService executorService) {
        return MoreExecutors.shutdownAndAwaitTermination(executorService, 10L, TimeUnit.SECONDS);
    }
}

展开全文 >>

【DIY无人机】电调固件升级[已收录]

2022-08-26

电调

为什么这里要说电调，因为电调对于多旋翼来说肯定是响应越快越好,多轴要的就是快速响应!但快速响应等高性能就意味着价格高,所以我们在价格和性能之间往往是不好选择的(就是因为穷,哈哈哈哈),但 BLHELI 电调开源固件解决了这个问题。我们只需要花 40-50 块钱买一个好盈天行者 20A/40A 电调，然后往电调里刷 BLHELI 的电调固件,然后你就得到了一个响应速度可以和 200-300 块钱的高价电调相当的廉价电调!话说回来这 BLHELI 固件也不是非刷不可,不刷就是响应慢了点,照样能飞起来!好盈天行者 20A 和好盈天行者 40A,选哪个都行,看你资金情况和长远考虑,其实好盈天行者 20A 就目前来说就够用了! 电调原固件可以,但响应不快!刷了 BLHeli 固件之后响应速度大大提高,这会让你四轴飞行更加稳定!