elastic-job集成到springboot教程，和它的一个异常处理办法:Sharding item parameters '1' format error, should be

Posted 2021-02-05 BigOcean

先说这个Sharding item parameters \'1\' format error, should be int=xx,int=xx异常吧，这是在做动态添加调度任务的时候出现的，网上找了一会没有搜到任何信息，最后发现，是添加任务这个方法里有一个漏洞。

这个源码出自：

 1 private ShardingItem parse(final String shardingItemParameter, final String originalShardingItemParameters) {
 2         String[] pair = shardingItemParameter.trim().split(KEY_VALUE_DELIMITER);
 3         if (2 != pair.length) {
 4             throw new JobConfigurationException("Sharding item parameters \'%s\' format error, should be int=xx,int=xx", originalShardingItemParameters);
 5         }
 6         try {
 7             return new ShardingItem(Integer.parseInt(pair[0].trim()), pair[1].trim());
 8         } catch (final NumberFormatException ex) {
 9             throw new JobConfigurationException("Sharding item parameters key \'%s\' is not an integer.", pair[0]);
10         }
11     }

 

修改前代码（报这个异常的代码）：

1  public void addJobScheduler(final Class<? extends SimpleJob> jobClass,
2       final String cron,
3       final int shardingTotalCount,
4       final String shardingItemParameters) {
5     JobCoreConfiguration coreConfig = JobCoreConfiguration.newBuilder(jobClass.getName(), cron, shardingTotalCount).shardingItemParameters(shardingItemParameters).build();
6     SimpleJobConfiguration simpleJobConfig = new SimpleJobConfiguration(coreConfig, jobClass.getCanonicalName());
7     JobScheduler jobScheduler = new JobScheduler(regCenter, LiteJobConfiguration.newBuilder(simpleJobConfig).build());
8     jobScheduler.init();
9   }

 

是不是发现不管你怎么设置，都给你报这个，你明明传的就不是1这个参数，还是给你报这个，问题出在build()那里，需要overwrite。修改后：

1  public void addJobScheduler(final Class<? extends SimpleJob> jobClass,
2       final String cron,
3       final int shardingTotalCount,
4       final String shardingItemParameters) {
5     JobCoreConfiguration coreConfig = JobCoreConfiguration.newBuilder(jobClass.getName(), cron, shardingTotalCount).shardingItemParameters(shardingItemParameters).build();
6     SimpleJobConfiguration simpleJobConfig = new SimpleJobConfiguration(coreConfig, jobClass.getCanonicalName());
7     JobScheduler jobScheduler = new JobScheduler(regCenter, LiteJobConfiguration.newBuilder(simpleJobConfig).overwrite(true).build());
8     jobScheduler.init();
9   }

红色代码为修改后加的代码。

 

先说说这个dangdang的elastic-job，它是一个分布式任务调度插件。今天我遇到的问题就是，有部分任务，在多节点环境中，不需要每个节点执行，比如只需要一个节点（确切地说就是作业分片总数=1）上运行的任务，这时候elastic-job就是个不错的选择，它可以很灵活的配置作业分片总数等。它的官方文档，链接指向配置说明：http://elasticjob.io/docs/elastic-job-lite/02-guide/config-manual/

 

那么spring boot中如何集成进它。需要的一个前提条件是zookeeper服务，这个一般项目里都会用到，你只需要连就好了，如果dev或者你们还没用上，可以找个教程安装一下。链接指向在 CentOS7 上安装 Zookeeper服务 。

然后.propertis配置文件（yml类同）：

regCenter.serverList=10.0.30.140:2181
regCenter.namespace=elastic-job

simpleJob.cron=0/5 * * * * ?  
# 作业分片总数,设为1只在一个节点执行
simpleJob.shardingTotalCount=1
simpleJob.shardingItemParameters=0=A,1=B,2=C

当然，pom中需要引入elastic-job：

<dependency>
    <artifactId>elastic-job-common-core</artifactId>
    <groupId>com.dangdang</groupId>
    <version>${elastic-job.version}</version>
  </dependency>
  <dependency>
    <artifactId>elastic-job-lite-core</artifactId>
    <groupId>com.dangdang</groupId>
    <version>${elastic-job.version}</version>
  </dependency>
  <dependency>
    <artifactId>elastic-job-lite-spring</artifactId>
    <groupId>com.dangdang</groupId>
    <version>${elastic-job.version}</version>
  </dependency>
  <dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-test</artifactId>
    <version>${curator.version}</version>
  </dependency>

本示例代码用到的版本是：

<elastic-job.version>2.1.5</elastic-job.version>
    <curator.version>2.10.0</curator.version>

连接zookeeper注册中心：

import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperConfiguration;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author binhy
 *@date 2019-1-22
 */
@Configuration
@ConditionalOnExpression("\'${regCenter.serverList}\'.length() > 0")
public class RegistryCenterConfig {

  @Bean(initMethod = "init")
  public ZookeeperRegistryCenter regCenter(@Value("${regCenter.serverList}") final String serverList, @Value("${regCenter.namespace}") final String namespace) {
    return new ZookeeperRegistryCenter(new ZookeeperConfiguration(serverList, namespace));
  }
}

然后做一个任务信息持久化：

import javax.annotation.Resource;
import javax.sql.DataSource;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import com.dangdang.ddframe.job.event.JobEventConfiguration;
import com.dangdang.ddframe.job.event.rdb.JobEventRdbConfiguration;

/**
 * @author binhy
 *@date 2019-1-22
 */
@Configuration
public class JobEventConfig {

  @Resource
  private DataSource dataSource;

  @Bean
  public JobEventConfiguration jobEventConfiguration() {
    return new JobEventRdbConfiguration(dataSource);
  }
}

当你最后运行后会发现你的库里多了两张表job_execution_log和job_status_trace_log他们会详细的记录你的任务执行信息，包括执行ip，开始结束时间等，还是非常不错的。

 

然后需要一个任务管理类，初始化一些任务，我这里把动态添加任务的方法也写在了这里。因为这种形式需要你一个任务写一个配置和类去实现。动态添加会省很多事。

import com.goopal.exdata.dangdang.DemoJob;
import javax.annotation.Resource;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.event.JobEventConfiguration;
import com.dangdang.ddframe.job.lite.api.JobScheduler;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import com.dangdang.ddframe.job.lite.spring.api.SpringJobScheduler;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;

/**
 * @author binhy
 * @date 2019-1-22
 */
@Configuration
public class SimpleJobConfig {

  @Resource
  private ZookeeperRegistryCenter regCenter;

  @Resource
  private JobEventConfiguration jobEventConfiguration;

  @Bean
  public SimpleJob simpleJob() {
    return new DemoJob();
  }

  @Bean(initMethod = "init")
  public JobScheduler simpleJobScheduler(final SimpleJob simpleJob, @Value("${simpleJob.cron}") final String cron, @Value("${simpleJob.shardingTotalCount}") final int shardingTotalCount,
      @Value("${simpleJob.shardingItemParameters}") final String shardingItemParameters) {
    return new SpringJobScheduler(simpleJob, regCenter, getLiteJobConfiguration(simpleJob.getClass(), cron, shardingTotalCount, shardingItemParameters), jobEventConfiguration);
  }

  private LiteJobConfiguration getLiteJobConfiguration(final Class<? extends SimpleJob> jobClass, final String cron, final int shardingTotalCount, final String shardingItemParameters) {
    return LiteJobConfiguration.newBuilder(new SimpleJobConfiguration(JobCoreConfiguration.newBuilder(
        jobClass.getName(), cron, shardingTotalCount).shardingItemParameters(shardingItemParameters).build(), jobClass.getCanonicalName())).overwrite(true).build();
  }

  /**
   * 动态添加
   * @param jobClass
   * @param cron
   * @param shardingTotalCount
   * @param shardingItemParameters
   */
  public void addJobScheduler(final Class<? extends SimpleJob> jobClass,
      final String cron,
      final int shardingTotalCount,
      final String shardingItemParameters) {
    JobCoreConfiguration coreConfig = JobCoreConfiguration.newBuilder(jobClass.getName(), cron, shardingTotalCount).shardingItemParameters(shardingItemParameters).build();
    SimpleJobConfiguration simpleJobConfig = new SimpleJobConfiguration(coreConfig, jobClass.getCanonicalName());
    JobScheduler jobScheduler = new JobScheduler(regCenter, LiteJobConfiguration.newBuilder(simpleJobConfig).overwrite(true).build());
    jobScheduler.init();
  }
}

到这里，你在配置文件中配置的定时任务就已经可以在多节点环境中，仅在1个节点执行了。需要添加更多的不同cron的任务，只需要在代码业务逻辑处调用即可。如：

 @Autowired
  private  SimpleJobConfig simpleJobConfig;
  @Override
  public void run(ApplicationArguments args) throws Exception {

    simpleJobConfig.addJobScheduler(AnalysisData.class,"0/3 * * * * ?",3,"0=A,1=B,2=C");//0=A,1=B,2=C
  }

 

如果有帮助到你，给我点个赞哦~