spring+quartzに基づく分散スケジュールタスクフレームワークの実装-＆＃＆チュートリアル-php.cn

問題の背景

当社は現在 200 名の従業員を擁する急成長中のスタートアップ企業で、主なビジネスはアプリケーションの反復と更新のサイクルが比較的速いです。反復のペースが増加し、システム全体の制御ができなくなりました

クラスターが存在する前は、会社のスケジュールされたタスクが実装されていました

初期の頃、アプリケーションへのアクセス数は 1 つのサーバーではそれほど多くありませんでした。十分に使用でき、スケジュールされたタスクを実行する必要があるアプリケーションもたくさんありました

クラスターを使用すると、会社のスケジュールされたタスクが実装されます

ユーザーの数が増えると、1 つのサーバーでは処理できないアクセス数も増加します。高い同時実行性の要件があるため、企業はアプリケーションをクラスター内にデプロイします。フロントエンドは nginx を介してプロキシされます (IP + ポート + アプリケーション名を使用した直接アクセスを避けるために、アプリケーションサーバーの IP はファイアウォールによって分離される場合があります)。

クラスター環境では、同じスケジュールされたタスクがクラスター内のすべてのマシンで実行されます。このように、スケジュールされたタスクが繰り返し実行されるため、サーバーの負荷が増大するだけでなく、追加の予期せぬイベントも発生します。スケジュールされたタスクが繰り返し実行されるためエラーが発生するため、同社の解決策は、クラスターの数に応じて、スケジュールされたタスク内のタスクをクラスター内の各マシンに均等に分散することです (ここでの平均スコアは、スケジュールされたタスクが最初に実行されたことを意味します)。 1 台のマシン上で、まずこのタスクを人為的にいくつかの部分に分割し、すべてのマシンでこの人を実行させます)

クラスター内でスケジュールされたタスクを実装する現在の方法の欠陥

同社がクラスター内でスケジュールされたタスクを処理する現在の方法は次のとおりです。これは分散処理方式ですが、疑似分散方式 (社内ではネイティブ方式として一般に知られています) であるため、この方式には明らかな欠陥があります。クラスター内のマシンがダウンすると、スケジュールされたタスク全体がハングアップします。または、一度に完了できない場合は、ビジネスに重大な影響を及ぼします

欠陥の解決策 (この記事の焦点)

関連情報を参照した後、spring+quartz を使用して、実際の分散スケジュールされたタスクシステムを構築します。 Quartz フレームワークがネイティブであることを学びました

分散スケジュールされたタスクをサポートする開発 IDE: Intellij IDEA

JDK バージョン: 1.8

Spring バージョン: 4.2.6

Quartz バージョン: 2.2.1

Spring と Quartz の統合構成

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"
    xsi:schemaLocation="http://www.springframework.org/schema/beans
      http://www.springframework.org/schema/beans/spring-beans-4.0.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
 
  <context:component-scan base-package="com.aaron.clusterquartz.job"/>
 
  <bean name="dataSource" class="org.springframework.jndi.JndiObjectFactoryBean">
    <!-- tomcat -->
    <!--<property name="jndiName" value="java:comp/env/jndi/mysql/quartz"/>-->
 
    <!-- jboss -->
    <property name="jndiName" value="jdbc/quartz"/>
  </bean>
  <!-- 分布式事务配置 start -->
 
  <!-- 配置线程池-->
  <bean name="executor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
    <property name="corePoolSize" value="15"/>
    <property name="maxPoolSize" value="25"/>
    <property name="queueCapacity" value="100"/>
  </bean>
 
  <bean name="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
    <property name="dataSource" ref="dataSource"/>
  </bean>
 
  <!-- 配置调度任务-->
  <bean name="quartzScheduler" class="org.springframework.scheduling.quartz.SchedulerFactoryBean">
    <property name="configLocation" value="classpath:quartz.properties"/>
    <property name="dataSource" ref="dataSource"/>
    <property name="transactionManager" ref="transactionManager"/>
 
    <!-- 任务唯一的名称，将会持久化到数据库-->
    <property name="schedulerName" value="baseScheduler"/>
 
    <!-- 每台集群机器部署应用的时候会更新触发器-->
    <property name="overwriteExistingJobs" value="true"/>
    <property name="applicationContextSchedulerContextKey" value="appli"/>
 
    <property name="jobFactory">
      <bean class="com.aaron.clusterquartz.autowired.AutowiringSpringBeanJobFactory"/>
    </property>
 
    <property name="triggers">
      <list>
        <ref bean="printCurrentTimeScheduler"/>
      </list>
    </property>
    <property name="jobDetails">
      <list>
        <ref bean="printCurrentTimeJobs"/>
      </list>
    </property>
 
    <property name="taskExecutor" ref="executor"/>
 
  </bean>
 
  <!-- 配置Job详情 -->
  <bean name="printCurrentTimeJobs" class="org.springframework.scheduling.quartz.JobDetailFactoryBean">
    <property name="jobClass" value="com.aaron.clusterquartz.job.PrintCurrentTimeJobs"/>
　　　　<!--因为我使用了spring的注解，所以这里可以不用配置scheduler的属性-->
    <!--<property name="jobDataAsMap">
      <map>
        <entry key="clusterQuartz" value="com.aaron.framework.clusterquartz.job.ClusterQuartz"/>
      </map>
    </property>-->
    <property name="durability" value="true"/>
    <property name="requestsRecovery" value="false"/>
  </bean>
 
  <!-- 配置触发时间 -->
  <bean name="printCurrentTimeScheduler" class="com.aaron.clusterquartz.cron.PersistableCronTriggerFactoryBean">
    <property name="jobDetail" ref="printCurrentTimeJobs"/>
    <property name="cronExpression">
      <value>0/10 * * * * ?</value>
    </property>
    <property name="timeZone">
      <value>GMT+8:00</value>
    </property>
  </bean>
 
  <!-- 分布式事务配置 end -->
</beans>

ログイン後にコピー

quartz プロパティファイル

#============================================================================
# Configure JobStore
# Using Spring datasource in quartzJobsConfig.xml
# Spring uses LocalDataSourceJobStore extension of JobStoreCMT
#============================================================================
org.quartz.jobStore.useProperties=true
org.quartz.jobStore.tablePrefix = QRTZ_
org.quartz.jobStore.isClustered = true
org.quartz.jobStore.clusterCheckinInterval = 5000
org.quartz.jobStore.misfireThreshold = 60000
org.quartz.jobStore.txIsolationLevelReadCommitted = true
 
# Change this to match your DB vendor
org.quartz.jobStore.class = org.quartz.impl.jdbcjobstore.JobStoreTX
org.quartz.jobStore.driverDelegateClass = org.quartz.impl.jdbcjobstore.StdJDBCDelegate
 
 
#============================================================================
# Configure Main Scheduler Properties
# Needed to manage cluster instances
#============================================================================
org.quartz.scheduler.instanceId=AUTO
org.quartz.scheduler.instanceName=MY_CLUSTERED_JOB_SCHEDULER
org.quartz.scheduler.rmi.export = false
org.quartz.scheduler.rmi.proxy = false
 
 
#============================================================================
# Configure ThreadPool
#============================================================================
org.quartz.threadPool.class = org.quartz.simpl.SimpleThreadPool
org.quartz.threadPool.threadCount = 10
org.quartz.threadPool.threadPriority = 5
org.quartz.threadPool.threadsInheritContextClassLoaderOfInitializingThread = true

ログイン後にコピー

関連クラスの説明

AutowiringSpringBeanJobFactory クラスは、スケジューラで Spring アノテーションを使用するため、アノテーションを使用しない場合、このクラスを適用して
SpringBeanJobFactory を使用することはできません

package com.aaron.clusterquartz.autowired;
 
import org.quartz.spi.TriggerFiredBundle;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.scheduling.quartz.SpringBeanJobFactory;
 
/**
 * @author 
 * @description 使job类支持spring的自动注入
 * @date 2016-05-27
 */
public class AutowiringSpringBeanJobFactory extends SpringBeanJobFactory implements ApplicationContextAware
{
  private transient AutowireCapableBeanFactory beanFactory;
 
  public void setApplicationContext(ApplicationContext applicationContext) throws BeansException
  {
    beanFactory = applicationContext.getAutowireCapableBeanFactory();
  }
 
 
  @Override
  protected Object createJobInstance(TriggerFiredBundle bundle) throws Exception
  {
    Object job = super.createJobInstance(bundle);
    beanFactory.autowireBean(job);
    return job;
  }
}

ログイン後にコピー

package com.aaron.clusterquartz.job;
 
import com.arron.util.DateUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.quartz.QuartzJobBean;
 
import java.util.Date;
 
/**
 * @author 
 * @description 一句话描述该文件的用途
 * @date 2016-05-23
 */
public class PrintCurrentTimeJobs extends QuartzJobBean
{
  private static final Log LOG_RECORD = LogFactory.getLog(PrintCurrentTimeJobs.class);
 
　　//这里就是因为有上文中的AutowiringSpringBeanJobFactory才可以使用@Autowired注解，否则只能在配置文件中设置这属性的值
  @Autowired
  private ClusterQuartz clusterQuartz;
 
 
  protected void executeInternal(JobExecutionContext jobExecutionContext) throws JobExecutionException
  {
    LOG_RECORD.info("begin to execute task," + DateUtils.dateToString(new Date()));
 
    clusterQuartz.printUserInfo();
 
    LOG_RECORD.info("end to execute task," + DateUtils.dateToString(new Date()));
 
  }
}

ログイン後にコピー

テスト結果:

コンピューターが 1 台しかないため、テストのために 2 つのポート 8080 と 8888 を開き、上記のスケジュールされたタスクを 10 秒ごとに実行するように設定しました。

ポート 8080 を起動すると、コンソールが 10 秒ごとにステートメントを出力することがわかります

spring+quartzに基づく分散スケジュールタスクフレームワークの実装