第四章 - 分布式定时任务框架ElasticJob之Spring框架整合DataflowJob作业

港控/mmm° 2023-10-03 18:51 23阅读 0赞

#### 分布式定时任务框架ElasticJob之Spring框架整合DataflowJob作业 ####

*  前言
 *  一、 开发环境准备
 *  二、使用步骤
 *   *  1.引入POM文件
     *  2.resource目录下引入spring-config.xml配制文件
     *  3.编写代码测试
 *  总结

--------------------

## 前言 ##

上一章节我们介绍了Spring框架整合SimpleJob作业来实现定时任务，本章节介绍Spring框架整合DataflowJob作业

--------------------

## 一、 开发环境准备 ##

**1.Eclipse**  
**2.zookeeper**  
**3.JDK1.8**  
**4.maven3.6.3**  
**5.mysql8.0**

## 二、使用步骤 ##

### 1.引入POM文件 ###

创建一个Maven项目，导入相关maven依赖，pom.xml文件如下：

<?xml version="1.0" encoding="UTF-8"?>
    
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
      <modelVersion>4.0.0</modelVersion>
    
      <groupId>com.example</groupId>
      <artifactId>spring-elasticjob</artifactId>
      <version>0.0.1-SNAPSHOT</version>
      <packaging>war</packaging>
    
      <name>spring-elasticjob Maven Webapp</name>
      
      <url>http://www.example.com</url>
    
      <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
      </properties>
    
      <dependencies>
        <dependency>
          <groupId>junit</groupId>
          <artifactId>junit</artifactId>
          <version>4.11</version>
          <scope>test</scope>
        </dependency>
        
    	<dependency>
    	    <groupId>org.springframework</groupId>
    	    <artifactId>spring-context</artifactId>
    	    <version>5.1.5.RELEASE</version>
    	</dependency>
        
    	<dependency>
    	    <groupId>org.springframework</groupId>
    	    <artifactId>spring-web</artifactId>
    	    <version>5.1.5.RELEASE</version>
    	</dependency>
        
    	<dependency>
    	    <groupId>com.dangdang</groupId>
    	    <artifactId>elastic-job-lite-core</artifactId>
    	    <version>2.1.5</version>
    	</dependency>
        
    	<dependency>
    	    <groupId>com.dangdang</groupId>
    	    <artifactId>elastic-job-lite-spring</artifactId>
    	    <version>2.1.5</version>
    	</dependency>
        
    	<dependency>
    	    <groupId>org.projectlombok</groupId>
    	    <artifactId>lombok</artifactId>
    	    <version>1.18.6</version>
    	    <scope>provided</scope>
    	</dependency>
    	<dependency>
    	 <groupId>org.slf4j</groupId>
    	 <artifactId>slf4j-log4j12</artifactId>
    	 <version>1.7.26</version>
    	</dependency>
    	
    	<dependency>
    	    <groupId>org.apache.commons</groupId>
    	    <artifactId>commons-dbcp2</artifactId>
    	    <version>2.6.0</version>
    	</dependency>
      </dependencies>
    
      <build>
        <finalName>spring-elasticjob</finalName> 
        <plugins>
        	<plugin>
            <groupId>org.apache.tomcat.maven</groupId>
            <artifactId>tomcat7-maven-plugin</artifactId>
            <version>2.2</version>
            <executions>
              <execution>
                <id>tomcat-run</id>
                <goals>
                  <goal>exec-war-only</goal>
                </goals>
                <phase>package</phase>
                <configuration>
                  <path>/</path>
                </configuration>
              </execution>
            </executions>
          </plugin>  	
        </plugins>
      </build>
    </project>

### 2.resource目录下引入spring-config.xml配制文件 ###

spring框架与javaApi整合作业的主要区别在于前者是通过配置文件的方式注册作业到注册中心的，下面附上配置文件：

<?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:reg="http://www.dangdang.com/schema/ddframe/reg"
           xmlns:job="http://www.dangdang.com/schema/ddframe/job"
           xsi:schemaLocation="http://www.springframework.org/schema/beans
           http://www.springframework.org/schema/beans/spring-beans.xsd
                http://www.dangdang.com/schema/ddframe/reg
                http://www.dangdang.com/schema/ddframe/reg/reg.xsd
                http://www.dangdang.com/schema/ddframe/job
                http://www.dangdang.com/schema/ddframe/job/job.xsd       
    ">
    	
        <reg:zookeeper server-lists="localhost:2181" base-sleep-time-milliseconds="" namespace="spring-elasticjob" id="zkCenter"/>
    
    	<bean id="dataSource" class="org.apache.commons.dbcp2.BasicDataSource">
    		<property name="username" value="root"/>
    		<property name="password" value="root"/>
    		<property name="url" value="jdbc:mysql://localhost:3306/dataflow"></property>
    	</bean>
       <job:dataflow registry-center-ref="zkCenter" cron="0/10 * * * * ?" sharding-total-count="2" id="myDataflowJob"
       class="com.example.job.MyDataflowJob" overwrite="true" event-trace-rdb-data-source="dataSource" streaming-process="true">
       		<job:listener class=""/>
       		<job:distribute-listener class="" started-time-out-milliseconds="" completed-timeout-milliseconds=""/>
       </job:dataflow>
    </beans>

### 3.编写代码测试 ###

package com.example.job;
    
    import java.util.ArrayList;
    import java.util.List;
    
    import com.dangdang.ddframe.job.api.ShardingContext;
    import com.dangdang.ddframe.job.api.dataflow.DataflowJob;
    
    public class MyDataflowJob implements DataflowJob<Integer>{
        
    
    	private List<Integer> list = new ArrayList();
    	{
        
    		list.add(0);
    		list.add(1);
    		list.add(2);
    		list.add(3);
    		list.add(4);
    		list.add(5);
    		list.add(6);
    		list.add(7);
    		list.add(8);
    		list.add(9);
    	}
    	
    	@Override
    	public List<Integer> fetchData(ShardingContext shardingContext) {
        
    		// TODO Auto-generated method stub
    		//数字 % 分片总数 == 当前分片项
    		List<Integer> rtnList = new ArrayList();
    		for(Integer index : list) {
        
    			if(index % shardingContext.getShardingTotalCount() == shardingContext.getShardingItem()) {
        
    				rtnList.add(index);
    				break;
    			}
    		}
    		try {
        
    			Thread.sleep(3000);
    		} catch (InterruptedException e) {
        
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    		System.out.println("我是分片项"+shardingContext.getShardingItem()+",我获取的数据是"+rtnList);
    		return rtnList;
    	}
    
    	@Override
    	public void processData(ShardingContext shardingContext, List<Integer> data) {
        
    		// TODO Auto-generated method stub
    		list.removeAll(data);
    		try {
        
    			Thread.sleep(5000);
    		} catch (InterruptedException e) {
        
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    		System.out.println("我是分片项"+shardingContext.getShardingItem() + ",我移除的数据是" + data);
    	}
    }

--------------------

## 总结 ##

本章介绍了Spring框架整合DataflowJob的基本使用，下章节介绍SpringBoot框架整合SimpleJob作业基本使用流程。