技术组件-任务分发工具

介绍:

在分布式系统中进行任务分发,分给各服务器进行分页处理!

原则:

轻量级,只管分发,不做调度等重型任务.----我只是个插件

代码路径:

https://gitee.com/kaiyang_taichi/allot_plugns.git

其中:

• allot-plugin为插件代码
• demo-test 中有测试使用例子

模型概念:

任务分发插件.png

任务工厂(JobAlloterFactory): 生产任务分发者对工厂

任务分发者(JobAlloter): 任务分发者

任务组(相同jobGroupName):对应一组任务,有相同的处理类

一次任务(相同job_id的多个job):对应一次任务分发

一个任务(一个job):对应一条任务数据,对应业务数据一页

使用方法(可参考上面说的demo-test项目):

1.引入pom:

<dependency>
            <groupId>cn.creditease.std</groupId>
            <artifactId>allot-plugin</artifactId>
            <version>1.0-SNAPSHOT</version>
            <exclusions>
                <exclusion>
                    <artifactId>spring-amqp</artifactId>
                    <groupId>org.springframework.amqp</groupId>
                </exclusion>
            </exclusions>
        </dependency>

2. 配置生产工厂(demo中的AlloterConfig类):

package com.example.demo.config;

import cn.creditease.std.config.JobConsumerConfig;
import cn.creditease.std.factory.JobAlloterFactory;
import cn.creditease.std.factory.JobAlloterFactoryBuilder;
import cn.creditease.std.utils.IpUtils;
import java.util.HashMap;
import java.util.Map;
import javax.sql.DataSource;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @Description:
 * @Author: kai.yang
 * @Date: 2019-08-13 10:11
 */
@Configuration
public class AlloterConfig {

    @Autowired
    RabbitTemplate rabbitTemplate;

    @Autowired
    ConnectionFactory connectionFactory;

    @Autowired
    DataSource dataSource;

    @Value("${server.port}")
    private int port;


    @Bean
    public JobAlloterFactory getJobAlloterManager() {
//  对应每个消费者的个性化配置,如果不配将采用统一默认值
        JobConsumerConfig jobConsumerConfig = new JobConsumerConfig().
                setMaxJobCount(40).setMaxConsumerCountByJobGroupPerServer(3);
        Map<String, JobConsumerConfig> jobConsumerConfigMap = new HashMap<String, JobConsumerConfig>();
        jobConsumerConfigMap.put("test", jobConsumerConfig);

        return new JobAlloterFactoryBuilder().setConnectionFactory(connectionFactory)
            .setConsumerId(IpUtils.getLocalIp() + ":" + port).setDataSource(dataSource).setJobConsumerConfigMap(jobConsumerConfigMap)
            .setRabbitTemplate(rabbitTemplate).buildMqFactory();

    }

}

3. 添加对应要处理任务的消费者(demo中的TestAllotProcessor类)

package com.example.demo.allot;

import cn.creditease.std.consumer.AbstactJobConsumeWorker;
import com.alibaba.fastjson.JSON;
import com.example.demo.entity.UserEntity;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

/**
 * @Description:
 * @Author: kai.yang
 * @Date: 2019-08-13 10:15
 */
@Component
public class TestAllotProcessor extends AbstactJobConsumeWorker<UserEntity> {

    private static final Logger logger = LoggerFactory.getLogger(TestAllotProcessor.class);

    public List<UserEntity> doReader(int pageNo, int pageSize) {
        logger.info("处理job数据{},{}", pageNo, pageSize);
        //模拟数据返回
        //从数据库读取pageNo页数据PageSize条
        List<UserEntity> userEntities = new ArrayList<UserEntity>();
        for (int i = pageNo * pageSize; i < (pageNo + 1) * pageSize; i++) {
            userEntities.add(new UserEntity(i, "开心" + i + "号", i));
        }

        return userEntities;
    }

    public boolean doProcess(List<UserEntity> datas) {
        logger.info("处理job业务数据{}", JSON.toJSONString(datas));
        for (UserEntity u : datas) {
            //模拟处理，将所有人成绩加10分
            u.setGrade(u.getGrade() + 10);
        }
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return true;
    }

    public String getJobName() {
        return "test";
    }
}

4. 在需要分发任务处进行任务分发,注意你要先算出一共有多少条业务数据,

package com.example.demo.controller;

import cn.creditease.std.JobAlloter;
import cn.creditease.std.factory.JobAlloterFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @Description:
 * @Author: kai.yang
 * @Date: 2019-08-13 10:00
 */
@RestController
@RequestMapping("/allot")
public class AllotTestController {

    @Autowired
    JobAlloterFactory jobAlloterManager;

    /**
     * 分发测试
     */
    @RequestMapping("test")
    public String test() {
        //模拟计算总条数有1000条数据
        int sum = 1000;
        String jobName = "test";
        int maxDuration = 1;
        JobAlloter alloter = jobAlloterManager.getAlloter(1000, jobName, maxDuration);
        alloter.start();
        return "ok";
    }


}

解释下:
sum: 总业务待处理数
maxDuration: 为设置的当前任务多长时间为超时时间,单位为分钟,如这个任务处于PROCESS状态10分钟了,你认为其为失败,那就配10.如果出现这种任务系统会自动补偿恢复.

jobName:对应joaGroupName,和你的处理其名字一一对应

5. 好了下面就正常使用了.目前仅支持mq作为通知机制.后续有需要可以配置redis、zookeepr等机制