Spring的任务调度处理

算法之名

发布于 2020-10-26 16:29:47

5230

发布于 2020-10-26 16:29:47

Spring封装了JDK的任务调度线程池和任务调用，并使用标签就可以开启一个任务调用。

先进行一个Spring的任务调度线程池的配置，此时是多线程执行任务，如果不配置则默认为单线程串行执行任务。

@Configuration
@EnableScheduling
@Slf4j
public class ScheduleConfiguration implements SchedulingConfigurer {
    @Override
    public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
        ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
        taskScheduler.setPoolSize(Runtime.getRuntime().availableProcessors() * 2);
        taskScheduler.initialize();
        log.info("ThreadPoolTaskScheduler init poolSize");
        taskRegistrar.setTaskScheduler(taskScheduler);
    }
}

不进行上述配置的话，需要将@EnableScheduling配置到Springboot主启动类上(一般这么配置，但其实可以配置到任意一个@Configuration标记的配置类上)

@SpringBootApplication
@EnableScheduling
public class RediscachingApplication {

   public static void main(String[] args) {
      SpringApplication.run(RediscachingApplication.class, args);
   }

}

但一个系统有多个任务执行的时候，最好使用多线程配置，这里暂时不牵扯分布式任务调度的问题。

现在我们来测试每隔10秒进行一次打印

@Component
@Slf4j
public class TestScheduler {
    @Scheduled(fixedDelay = 1000 * 10)
    public void print() {
        log.info("测试打印");
    }
}

这种设置时间会等方法体执行完的第10秒开始执行，比如print()在第0秒开始执行，而print()方法本身执行了12秒，则下一次执行会在第22秒。

@Component
@Slf4j
public class TestScheduler {
    @Scheduled(fixedRate = 1000 * 10)
    public void print() {
        log.info("测试打印");
    }
}

这种设置当方法的执行时间超过任务调度频率时，调度器会在当前方法执行完成后立即执行下次任务。比如print()方法在第0秒开始执行，方法执行了12秒，那么下一次执行work()方法的时间是第12秒。

启动运行后，日志如下

2020-10-14 06:19:37.137  INFO 683 --- [TaskScheduler-1] c.g.r.scheduler.TestScheduler            : 测试打印
2020-10-14 06:19:47.141  INFO 683 --- [TaskScheduler-2] c.g.r.scheduler.TestScheduler            : 测试打印
2020-10-14 06:19:57.144  INFO 683 --- [TaskScheduler-1] c.g.r.scheduler.TestScheduler            : 测试打印
2020-10-14 06:20:07.146  INFO 683 --- [TaskScheduler-3] c.g.r.scheduler.TestScheduler            : 测试打印

我们可以看到它是由不同的线程来执行的。

当然也可以使用Cron表达式来设置

常用表达式

@Component
@Slf4j
public class TestScheduler {
    @Scheduled(cron = "0/10 * * * * *")
    public void print() {
        log.info("测试打印");
    }
}

这么写也是每隔10秒打印一次。

现在我们来写一个最简单的分布式调度，使用nacos

pom

<dependency>
   <groupId>com.alibaba.cloud</groupId>
   <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
</dependency>

<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-aop</artifactId>
</dependency>

<dependencyManagement>
   <dependencies>
      <dependency>
         <groupId>org.springframework.cloud</groupId>
         <artifactId>spring-cloud-dependencies</artifactId>
         <version>Greenwich.SR2</version>
         <type>pom</type>
         <scope>import</scope>
      </dependency>
      <dependency>
         <groupId>com.alibaba.cloud</groupId>
         <artifactId>spring-cloud-alibaba-dependencies</artifactId>
         <version>2.1.1.RELEASE</version>
         <type>pom</type>
         <scope>import</scope>
      </dependency>
   </dependencies>
</dependencyManagement>

配置文件

spring:
  cloud:
    nacos:
      discovery:
        server-addr: 127.0.0.1:8848
  application:
      name: redis-caching

server:
  port: 8080

写一个标签

@Target({ ElementType.METHOD })
@Retention(RetentionPolicy.RUNTIME)
public @interface Scheduler {
}

一个AOP类

@Aspect
@Component
public class SchedulerAop {
    @Autowired
    private DiscoveryClient discoveryClient;
    @Value("${server.port}")
    private int port;
    @Value("${spring.application.name}")
    private String appName;

    @Around(value = "@annotation(com.guanjian.rediscaching.annotation.Scheduler)")
    public Object scheduler(ProceedingJoinPoint joinPoint) throws Throwable {
        List<ServiceInstance> nacos = discoveryClient.getInstances(appName);
        if (nacos != null && nacos.size() > 0) {
            String ip = IpUtils.getHostIp();
            if ((nacos.get(0).getHost() + nacos.get(0).getPort()).equals(ip + port)) {
                Object res = joinPoint.proceed();
                return res;
            }
        }
        return null;
    }
}

其中IpUtils的代码如下

@Slf4j
public class IpUtils {

    public static String getHostIp() {
        String ip = null;
        try {
            //枚举本机所有的网络接口
            Enumeration<NetworkInterface> en = NetworkInterface
                    .getNetworkInterfaces();
            while (en.hasMoreElements()) {
                NetworkInterface intf = (NetworkInterface) en.nextElement();
                //遍历所有Ip
                Enumeration<InetAddress> enumIpAddr = intf.getInetAddresses();
                while (enumIpAddr.hasMoreElements()) {
                    InetAddress inetAddress = (InetAddress) enumIpAddr
                            .nextElement();
                    //获取类似192.168的内网IP
                    if (!inetAddress.isLoopbackAddress()  //isLoopbackAddress()是否是本机的IP地址(127开头的,一般指127.0.0.1)
                            && !inetAddress.isLinkLocalAddress()  //isLinkLocalAddress()是否是本地连接地址(任意开头)
                            && inetAddress.isSiteLocalAddress()) {  //isSiteLocalAddress()是否是地区本地地址(192.168段或其他内网IP)
                        ip = inetAddress.getHostAddress();
                    }
                }
            }
        } catch (SocketException e) {
            log.error("Fail to get IP address.", e);
        }
        return ip;
    }

    public static String getHostName() {
        String hostName = null;
        try {
            Enumeration<NetworkInterface> en = NetworkInterface
                    .getNetworkInterfaces();
            while (en.hasMoreElements()) {
                NetworkInterface intf = (NetworkInterface) en.nextElement();
                Enumeration<InetAddress> enumIpAddr = intf.getInetAddresses();
                while (enumIpAddr.hasMoreElements()) {
                    InetAddress inetAddress = (InetAddress) enumIpAddr
                            .nextElement();
                    if (!inetAddress.isLoopbackAddress()
                            && !inetAddress.isLinkLocalAddress()
                            && inetAddress.isSiteLocalAddress()) {
                        hostName = inetAddress.getHostName();
                    }
                }
            }
        } catch (SocketException e) {
            log.error("Fail to get host name.", e);
        }
        return hostName;
    }
}

最后依然是这个测试类，打上标签

@Component
@Slf4j
public class TestScheduler {
    @Scheduler
    @Scheduled(fixedRate = 1000 * 10)
    public void print() {
        log.info("测试打印");
    }
}

现在我们来启动第一个进程

nacos注册中心注册了该实例

日志中也开始进行打印

2020-10-14 11:10:46.941  INFO 648 --- [TaskScheduler-1] c.g.r.scheduler.TestScheduler            : 测试打印
2020-10-14 11:10:51.812  INFO 648 --- [TaskScheduler-1] c.g.r.scheduler.TestScheduler            : 测试打印
2020-10-14 11:11:01.811  INFO 648 --- [TaskScheduler-2] c.g.r.scheduler.TestScheduler            : 测试打印
2020-10-14 11:11:11.813  INFO 648 --- [TaskScheduler-1] c.g.r.scheduler.TestScheduler            : 测试打印
2020-10-14 11:11:21.811  INFO 648 --- [TaskScheduler-3] c.g.r.scheduler.TestScheduler            : 测试打印
2020-10-14 11:11:31.811  INFO 648 --- [TaskScheduler-2] c.g.r.scheduler.TestScheduler            : 测试打印
2020-10-14 11:11:41.812  INFO 648 --- [TaskScheduler-4] c.g.r.scheduler.TestScheduler            : 测试打印
2020-10-14 11:11:51.812  INFO 648 --- [TaskScheduler-1] c.g.r.scheduler.TestScheduler            : 测试打印

现在我们修改端口，启动第二个进程

server:
  port: 8081

启动成功后

我们可以看到实例数变成了2

后台打印

我们可以看到第一个进程的后台日志停止了打印，而第二个进程的后台日志开始打印

2020-10-14 11:15:23.925  INFO 693 --- [TaskScheduler-6] c.g.r.scheduler.TestScheduler            : 测试打印
2020-10-14 11:15:33.922  INFO 693 --- [TaskScheduler-2] c.g.r.scheduler.TestScheduler            : 测试打印
2020-10-14 11:15:43.925  INFO 693 --- [TaskScheduler-7] c.g.r.scheduler.TestScheduler            : 测试打印
2020-10-14 11:15:53.926  INFO 693 --- [TaskScheduler-4] c.g.r.scheduler.TestScheduler            : 测试打印
2020-10-14 11:16:03.925  INFO 693 --- [TaskScheduler-8] c.g.r.scheduler.TestScheduler            : 测试打印
2020-10-14 11:16:13.927  INFO 693 --- [TaskScheduler-1] c.g.r.scheduler.TestScheduler            : 测试打印

当然这是不一定的，两个进程谁打印谁不打印都是随机的，但可以肯定的是，只有一个进程可以打印日志，另外一个进程则不会做出打印操作。

如果我们结束打印日志的这个进程，则另外一个进程就会开始打印日志。

本文参与腾讯云自媒体分享计划，分享自作者个人站点/博客。

如有侵权请联系 cloudcommunity@tencent.com 删除

编程算法