介紹:

在分布式系統(tǒng)中進行任務(wù)分發(fā),分給各服務(wù)器進行分頁處理!

原則:

輕量級,只管分發(fā),不做調(diào)度等重型任務(wù).----我只是個插件

代碼路徑:

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

其中:

• allot-plugin為插件代碼
• demo-test 中有測試使用例子

模型概念:

任務(wù)分發(fā)插件.png

任務(wù)工廠(JobAlloterFactory): 生產(chǎn)任務(wù)分發(fā)者對工廠

任務(wù)分發(fā)者(JobAlloter): 任務(wù)分發(fā)者

任務(wù)組(相同jobGroupName):對應(yīng)一組任務(wù),有相同的處理類

一次任務(wù)(相同job_id的多個job):對應(yīng)一次任務(wù)分發(fā)

一個任務(wù)(一個job):對應(yīng)一條任務(wù)數(shù)據(jù),對應(yīng)業(yè)務(wù)數(shù)據(jù)一頁

使用方法(可參考上面說的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. 配置生產(chǎn)工廠(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() {
//  對應(yīng)每個消費者的個性化配置,如果不配將采用統(tǒng)一默認值
        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. 添加對應(yīng)要處理任務(wù)的消費者(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數(shù)據(jù){},{}", pageNo, pageSize);
        //模擬數(shù)據(jù)返回
        //從數(shù)據(jù)庫讀取pageNo頁數(shù)據(jù)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業(yè)務(wù)數(shù)據(jù){}", 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. 在需要分發(fā)任務(wù)處進行任務(wù)分發(fā),注意你要先算出一共有多少條業(yè)務(wù)數(shù)據(jù),

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;

    /**
     * 分發(fā)測試
     */
    @RequestMapping("test")
    public String test() {
        //模擬計算總條數(shù)有1000條數(shù)據(jù)
        int sum = 1000;
        String jobName = "test";
        int maxDuration = 1;
        JobAlloter alloter = jobAlloterManager.getAlloter(1000, jobName, maxDuration);
        alloter.start();
        return "ok";
    }


}

解釋下:
sum: 總業(yè)務(wù)待處理數(shù)
maxDuration: 為設(shè)置的當前任務(wù)多長時間為超時時間,單位為分鐘,如這個任務(wù)處于PROCESS狀態(tài)10分鐘了,你認為其為失敗,那就配10.如果出現(xiàn)這種任務(wù)系統(tǒng)會自動補償恢復(fù).

jobName:對應(yīng)joaGroupName,和你的處理其名字一一對應(yīng)

5. 好了下面就正常使用了.目前僅支持mq作為通知機制.后續(xù)有需要可以配置redis、zookeepr等機制
   ()