对电子商务网站与建设的心得,辽宁城乡建设招投标交易平台,3d 代做网站,长春企业宣传片制作公司一、调度中心管理注册信息
1.JobApiController
执行器调用调度中心的url来实现注册、下线、回调等操作#xff1b;其主要的实现类是JobApiController#xff0c;调用/api/registry接口注册执行器信息#xff0c;调用/api/registryRemove接口下线执行器信息#xff0c;调用…一、调度中心管理注册信息
1.JobApiController
执行器调用调度中心的url来实现注册、下线、回调等操作其主要的实现类是JobApiController调用/api/registry接口注册执行器信息调用/api/registryRemove接口下线执行器信息调用/api/callback接口执行回调操作。
Controller
RequestMapping(/api)
public class JobApiController {Resourceprivate AdminBiz adminBiz;/*** api** param uri* param data* return*/RequestMapping(/{uri})ResponseBodyPermissionLimit(limitfalse)public ReturnTString api(HttpServletRequest request, PathVariable(uri) String uri, RequestBody(required false) String data) {// validif (!POST.equalsIgnoreCase(request.getMethod())) {return new ReturnTString(ReturnT.FAIL_CODE, invalid request, HttpMethod not support.);}if (urinull || uri.trim().length()0) {return new ReturnTString(ReturnT.FAIL_CODE, invalid request, uri-mapping empty.);}if (XxlJobAdminConfig.getAdminConfig().getAccessToken()!null XxlJobAdminConfig.getAdminConfig().getAccessToken().trim().length()0 !XxlJobAdminConfig.getAdminConfig().getAccessToken().equals(request.getHeader(XxlJobRemotingUtil.XXL_JOB_ACCESS_TOKEN))) {return new ReturnTString(ReturnT.FAIL_CODE, The access token is wrong.);}// services mappingif (callback.equals(uri)) {ListHandleCallbackParam callbackParamList GsonTool.fromJson(data, List.class, HandleCallbackParam.class);return adminBiz.callback(callbackParamList);} else if (registry.equals(uri)) {RegistryParam registryParam GsonTool.fromJson(data, RegistryParam.class);return adminBiz.registry(registryParam);} else if (registryRemove.equals(uri)) {RegistryParam registryParam GsonTool.fromJson(data, RegistryParam.class);return adminBiz.registryRemove(registryParam);} else {return new ReturnTString(ReturnT.FAIL_CODE, invalid request, uri-mapping( uri ) not found.);}}}2.AdminBizImpl
执行adminBiz.registry(registryParam)是调用实现类AdminBizImpl在实现类中调用registry(registryParam)来实现。
Service
public class AdminBizImpl implements AdminBiz {Overridepublic ReturnTString callback(ListHandleCallbackParam callbackParamList) {return JobCompleteHelper.getInstance().callback(callbackParamList);}Overridepublic ReturnTString registry(RegistryParam registryParam) {return JobRegistryHelper.getInstance().registry(registryParam);}Overridepublic ReturnTString registryRemove(RegistryParam registryParam) {return JobRegistryHelper.getInstance().registryRemove(registryParam);}}3.JobRegistryHelper
在AdminBizImpl中的实现也不难理解通过在初始化start()方法中创建的registryOrRemoveThreadPool线程池中执行异步注册任务注册信息写入到数据表xxl_job_registry中。
//AdminBizImpl.javapublic void start(){// for registry or removeregistryOrRemoveThreadPool new ThreadPoolExecutor(2,10,30L,TimeUnit.SECONDS,new LinkedBlockingQueueRunnable(2000),new ThreadFactory() {Overridepublic Thread newThread(Runnable r) {return new Thread(r, xxl-job, admin JobRegistryMonitorHelper-registryOrRemoveThreadPool- r.hashCode());}},new RejectedExecutionHandler() {Overridepublic void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {r.run();logger.warn( xxl-job, registry or remove too fast, match threadpool rejected handler(run now).);}});//...省略public ReturnTString registry(RegistryParam registryParam) {// validif (!StringUtils.hasText(registryParam.getRegistryGroup())|| !StringUtils.hasText(registryParam.getRegistryKey())|| !StringUtils.hasText(registryParam.getRegistryValue())) {return new ReturnTString(ReturnT.FAIL_CODE, Illegal Argument.);}// async execute//my-异步执行将注册信息持久化到数据库registryOrRemoveThreadPool.execute(new Runnable() {Overridepublic void run() {//my-写入数据库int ret XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().registryUpdate(registryParam.getRegistryGroup(), registryParam.getRegistryKey(), registryParam.getRegistryValue(), new Date());if (ret 1) {XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().registrySave(registryParam.getRegistryGroup(), registryParam.getRegistryKey(), registryParam.getRegistryValue(), new Date());// freshfreshGroupRegistryInfo(registryParam);}}});return ReturnT.SUCCESS;
}4.总结
执行器通过restful api形式注册到调度中心来调度中心JobApiController对应有3个注册、下线和回调的方法实现通过AdminBizImpl的adminBiz.registry(registryParam)来实际执行注册方法实际使用JobRegistryHelper类在JobRegistryHelper类中在初始化的时候会创建一个线程池每次注册执行器的时候会创建一个异步线程来将注册信息持久化的数据库 二、调度中心的配置和启动
1.添加权限控制
制定权限注解PermissionLimit其实现的逻辑在PermissionInterceptor中首先判断是否需要鉴权如果需要则根据cookie中拿到的用户信息查库判断是否有权限登录如果没有权限则重定向到登录页面或提示没有权限。
Target(ElementType.METHOD)
Retention(RetentionPolicy.RUNTIME)
public interface PermissionLimit {/*** 登录拦截 (默认拦截)*/boolean limit() default true;/*** 要求管理员权限** return*/boolean adminuser() default false;}Component
public class PermissionInterceptor implements AsyncHandlerInterceptor {Resourceprivate LoginService loginService;Overridepublic boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {//my-处理登录权限的逻辑if (!(handler instanceof HandlerMethod)) {return true; // proceed with the next interceptor}// if need loginboolean needLogin true;boolean needAdminuser false;HandlerMethod method (HandlerMethod)handler;PermissionLimit permission method.getMethodAnnotation(PermissionLimit.class);if (permission!null) {needLogin permission.limit();needAdminuser permission.adminuser();}if (needLogin) {XxlJobUser loginUser loginService.ifLogin(request, response);if (loginUser null) {response.setStatus(302);response.setHeader(location, request.getContextPath()/toLogin);return false;}if (needAdminuser loginUser.getRole()!1) {throw new RuntimeException(I18nUtil.getString(system_permission_limit));}request.setAttribute(LoginService.LOGIN_IDENTITY_KEY, loginUser);}return true; // proceed with the next interceptor}}将PermissionInterceptor添加到web配置文件中。
Configuration
public class WebMvcConfig implements WebMvcConfigurer {Resourceprivate PermissionInterceptor permissionInterceptor;Resourceprivate CookieInterceptor cookieInterceptor;Overridepublic void addInterceptors(InterceptorRegistry registry) {registry.addInterceptor(permissionInterceptor).addPathPatterns(/**);registry.addInterceptor(cookieInterceptor).addPathPatterns(/**);}}2.配置中心初始化
xxljob的初始化和销毁动作在XxlJobAdminConfig中配置完成。
Component
public class XxlJobAdminConfig implements InitializingBean, DisposableBean {private static XxlJobAdminConfig adminConfig null;public static XxlJobAdminConfig getAdminConfig() {return adminConfig;}// ---------------------- XxlJobScheduler ----------------------private XxlJobScheduler xxlJobScheduler;Overridepublic void afterPropertiesSet() throws Exception {adminConfig this;xxlJobScheduler new XxlJobScheduler();xxlJobScheduler.init();}Overridepublic void destroy() throws Exception {xxlJobScheduler.destroy();}
}具体初始化的操作。
public class XxlJobScheduler {private static final Logger logger LoggerFactory.getLogger(XxlJobScheduler.class);public void init() throws Exception {// init i18ninitI18n();// admin trigger pool startJobTriggerPoolHelper.toStart();// admin registry monitor runJobRegistryHelper.getInstance().start();// admin fail-monitor runJobFailMonitorHelper.getInstance().start();// admin lose-monitor run ( depend on JobTriggerPoolHelper )JobCompleteHelper.getInstance().start();// admin log report startJobLogReportHelper.getInstance().start();// start-schedule ( depend on JobTriggerPoolHelper )JobScheduleHelper.getInstance().start();logger.info( init xxl-job admin success.);}public void destroy() throws Exception {// stop-scheduleJobScheduleHelper.getInstance().toStop();// admin log report stopJobLogReportHelper.getInstance().toStop();// admin lose-monitor stopJobCompleteHelper.getInstance().toStop();// admin fail-monitor stopJobFailMonitorHelper.getInstance().toStop();// admin registry stopJobRegistryHelper.getInstance().toStop();// admin trigger pool stopJobTriggerPoolHelper.toStop();}// ---------------------- I18n ----------------------private void initI18n(){for (ExecutorBlockStrategyEnum item:ExecutorBlockStrategyEnum.values()) {item.setTitle(I18nUtil.getString(jobconf_block_.concat(item.name())));}}
