Elastic Job 定时任务配置与实现

一、配置Zookeeper注册中心

首先，需要配置Zookeeper作为分布式协调服务注册中心。下载相应的Zookeeper包并解压后，运行bin目录下的start命令即可启动Zookeeper服务。完成后，可通过ZooKeeper客户端工具（如zooinspector）连接到注册中心，完成后续配置。

二、添加Maven依赖

在项目的POM文件中添加以下依赖，确保能够正确引用Elastic Job Lite相关组件：


       
    
     com.dangdang
        
    
     elastic-job-lite-core
        
    
     2.1.5

三、编写定时任务业务逻辑

创建定时任务处理类ArchivieJob，实现业务逻辑处理。以下是核心代码逻辑：

import com.dangdang.ddframe.job.api.ShardingContext;import com.dangdang.ddframe.job.api.simple.SimpleJob;import java.util.List;import java.util.Map;public class ArchivieJob implements SimpleJob {    @Override    public void execute(ShardingContext shardingContext) {        int shardingItem = shardingContext.getShardingItem();        String shardingParameter = shardingContext.getShardingParameter();        // 从resume表中查询未归档记录        List
    
     > maps = JdbcUtil.executeQuery(            "SELECT * FROM resume WHERE state = '未归档' AND education = '" + shardingParameter + "' LIMIT 1;"        );        if (maps == null) {            System.out.println("数据处理完毕！！！！");            return;        }        // 更新记录状态为已归档        Map
     
       stringObjectMap = maps.get(0);        Object id = stringObjectMap.get("id");        Object name = stringObjectMap.get("name");        Object education = stringObjectMap.get("education");        System.out.println("=====>>>>>" + id + "," + name + "," + education);        // 插入到resume_bak表中        JdbcUtil.executeUpdate(            "UPDATE resume SET state = '已归档' WHERE id = ?",             id        );        JdbcUtil.executeUpdate(            "INSERT INTO resume_bak SELECT * FROM resume WHERE id = ?",             id        );    }}

四、启动定时任务调度

在ElasicJobMain类中配置任务调度：

import com.dangdang.ddframe.job.config.JobCoreConfiguration;import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;import com.dangdang.ddframe.job.lite.api.JobScheduler;import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperConfiguration;import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;public class ElasicJobMain {    public static void main(String[] args) {        // 配置Zookeeper注册中心        ZookeeperConfiguration zookeeperConfiguration = new ZookeeperConfiguration("192.168.1.6:2181", "data-archive-job");        ZookeeperRegistryCenter zookeeperRegistryCenter = new ZookeeperRegistryCenter(zookeeperConfiguration);        zookeeperRegistryCenter.init();        // 配置任务        JobCoreConfiguration jobCoreConfiguration = JobCoreConfiguration.newBuilder("archive-job", "*/2 * * * * ?", 3)            .shardingItemParameters("0=doctor,1=bachelor,2=master")            .build();        SimpleJobConfiguration simpleJobConfiguration = new SimpleJobConfiguration(jobCoreConfiguration, ArchivieJob.class.getName());        JobScheduler jobScheduler = new JobScheduler(zookeeperRegistryCenter, LiteJobConfiguration.newBuilder(simpleJobConfiguration).overwrite(true).build());        jobScheduler.init();    }}

五、JDBC辅助工具类

提供数据库操作功能，包含连接数据库、执行查询和更新操作等功能：

import java.sql.*;import java.util.ArrayList;import java.util.HashMap;import java.util.List;import java.util.Map;public class JdbcUtil {    private static String url = "jdbc:mysql://localhost:3306/bank?characterEncoding=utf8&useSSL=false";    private static String user = "root";    private static String password = "123456";    private static String driver = "com.mysql.jdbc.Driver";    static {        try {            Class.forName(driver);        } catch (ClassNotFoundException e) {            e.printStackTrace();        }    }    public static Connection getConnection() {        try {            return DriverManager.getConnection(url, user, password);        } catch (SQLException e) {            e.printStackTrace();        }        return null;    }    public static void close(ResultSet rs, PreparedStatement ps, Connection con) {        if (rs != null) {            try {                rs.close();            } catch (SQLException e) {                e.printStackTrace();            } finally {                if (ps != null) {                    try {                        ps.close();                    } catch (SQLException e) {                        e.printStackTrace();                    } finally {                        if (con != null) {                            try {                                con.close();                            } catch (SQLException e) {                                e.printStackTrace();                            }                        }                    }                }            }        }    }    public static void executeUpdate(String sql, Object... obj) {        Connection con = getConnection();        PreparedStatement ps = null;        try {            ps = con.prepareStatement(sql);            for (int i = 0; i < obj.length; i++) {                ps.setObject(i + 1, obj[i]);            }            ps.executeUpdate();        } catch (SQLException e) {            e.printStackTrace();        } finally {            close(null, ps, con);        }    }    public static List
    
     > executeQuery(String sql, Object... obj) {        Connection con = getConnection();        ResultSet rs = null;        PreparedStatement ps = null;        try {            ps = con.prepareStatement(sql);            for (int i = 0; i < obj.length; i++) {                ps.setObject(i + 1, obj[i]);            }            rs = ps.executeQuery();            List
     
      
       > list = new ArrayList<>();            int count = rs.getMetaData().getColumnCount();            while (rs.next()) {                Map
       
         map = new HashMap<>();                for (int i = 0; i < count; i++) {                    Object ob = rs.getObject(i + 1);                    String key = rs.getMetaData().getColumnName(i + 1);                    map.put(key, ob);                }                list.add(map);            }            return list;        } catch (SQLException e) {            e.printStackTrace();        } finally {            close(rs, ps, con);        }        return null;    }}

转载地址：http://tduq.baihongyu.com/

你可能感兴趣的文章

Python3：setup.py：完成所有任务的PIP安装(Build_ext+Install)