介绍:
在分布式系统中进行任务分发,分给各服务器进行分页处理!
原则:
轻量级,只管分发,不做调度等重型任务.----我只是个插件
代码路径:
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>
- 配置生产工厂(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();
}
}
- 添加对应要处理任务的消费者(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";
}
}
- 在需要分发任务处进行任务分发,注意你要先算出一共有多少条业务数据,
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,和你的处理其名字一一对应
- 好了下面就正常使用了.目前仅支持mq作为通知机制.后续有需要可以配置redis、zookeepr等机制
()
