函数之异步执行

总体介绍

异步任务是非常常见的一种开发模式,它在分布式的开发模式中有很多应用场景如:

  • 高并发场景中,我们一般采用把长流程切短,用异步方式去掉可以异步的非关键功能,缩小主流程响应时间,提升用户体验
  • 异构系统的集成调用,通过异步任务完成解耦与自动重试
  • 分布式系统最终一致性的可选方案

本文将介绍oinone是如何结合Spring+TbSchedule来完成异步任务

构建第一个异步任务

新建PetShopService和PetShopServiceImpl

1、 新建PetShopService定义updatePetShops方法

package pro.shushi.pamirs.demo.api.service;

import pro.shushi.pamirs.demo.api.model.PetShop;
import pro.shushi.pamirs.meta.annotation.Fun;
import pro.shushi.pamirs.meta.annotation.Function;

import java.util.List;

@Fun(PetShopService.FUN_NAMESPACE)
public interface PetShopService {
    String FUN_NAMESPACE = "demo.PetShop.PetShopService";

    @Function
    void updatePetShops(List<PetShop> petShops);
}

2、PetShopServiceImpl实现PetShopService接口并在updatePetShops增加@XAsync注解

package pro.shushi.pamirs.demo.core.service;

import org.springframework.stereotype.Component;
import pro.shushi.pamirs.demo.api.model.PetShop;
import pro.shushi.pamirs.demo.api.service.PetShopService;
import pro.shushi.pamirs.meta.annotation.Fun;
import pro.shushi.pamirs.meta.annotation.Function;
import pro.shushi.pamirs.trigger.annotation.XAsync;
import java.util.List;

@Fun(PetShopService.FUN_NAMESPACE)
@Component
public class PetShopServiceImpl implements PetShopService {
    @Override
    @Function
    @XAsync(displayName = "异步批量更新宠物商店",limitRetryNumber = 3,nextRetryTimeValue = 60)
    public void updatePetShops(List<PetShop> petShops) {
        new PetShop().updateBatch(petShops);
    }
}

a. displayName = "异步批量更新宠物商店",定义异步任务展示名称
b. limitRetryNumber = 3,定义任务失败重试次数,,默认:-1不断重试
c. nextRetryTimeValue = 60,定义任务失败重试的时间数,默认:3
d. nextRetryTimeUnit,定义任务失败重试的时间单位,默认:TimeUnitEnum.SECOND
e. delayTime,定义任务延迟执行的时间数,默认:0
f. delayTimeUnit,定义任务延迟执行的时间单位,默认:TimeUnitEnum.SECOND

修改PetShopBatchUpdateAction调用异步任务

  1. 引入PetShopService
  2. 修改conform方法,调用petShopService.updatePetShops方法

    package pro.shushi.pamirs.demo.core.action;
    @Model.model(PetShopBatchUpdate.MODEL_MODEL)
    @Component
    public class PetShopBatchUpdateAction {
    @Autowired
    private PetShopService petShopService;
    
    @Action(displayName = "确定",bindingType = ViewTypeEnum.FORM,contextType = ActionContextTypeEnum.SINGLE)
    public PetShopBatchUpdate conform(PetShopBatchUpdate data){
            List<PetShop> shops = ArgUtils.convert(PetShopProxy.MODEL_MODEL, PetShop.MODEL_MODEL,proxyList);
           // 调用异步任务
            petShopService.updatePetShops(shops);
        });
        return data;
    }
    }

不同应用如何隔离执行单元

在schedule跟模块部署一起的时候,多模块独立boot的情况下,需要做必要的配置。如果schedule独立部署则没有必要,因为全部走远程,不存在类找不到的问题

  1. 通过配置pamirs.zookeeper.rootPath,确保两组机器都能覆盖所有任务分片,这样不会漏数据
  2. 通过pamirs.event.schedule.ownSign来隔离。确保两组机器只取各自产生的数据,这样不会重复执行数据
    pamirs:
    zookeeper:
    zkConnectString: 127.0.0.1:2181
    zkSessionTimeout: 60000
    rootPath: /demo
    event:
    enabled: true
    schedule:
      enabled: true
      ownSign: demo
    rocket-mq:
      namesrv-addr: 127.0.0.1:9876

Oinone社区 作者:望闲原创文章,如若转载,请注明出处:https://doc.oinone.top/backend/11480.html

访问Oinone官网:https://www.oinone.top获取数式Oinone低代码应用平台体验

Like (0)
望闲's avatar望闲数式管理员
Previous 2024年5月25日 pm8:10
Next 2024年5月27日 pm8:19

相关推荐

  • 项目中排除掉特定的Hook和扩展点

    总体介绍 在共库共Redis的情况下,某些场景存在需要过滤掉特定Hook和扩展点(extpoint)的情况。本文介绍排除掉的配置方法 1. Oinone如何排除特定的Hook 配置: pamirs: framework: hook: excludes: – 排除的扩展点列表 示例: pamirs: framework: hook: excludes: – pro.shushi.pamirs.timezone.hook.TimezoneHookBefore – pro.shushi.pamirs.timezone.hook.TimezoneHookAfter – pro.shushi.pamirs.timezone.hook.TimezoneSessionInitHook – pro.shushi.pamirs.translate.hook.TranslateAfterHook 2. Oinone如何排除特定的扩展点 配置 pamirs: framework: extpoint: excludes: – 排除的扩展点列表 示例: pamirs: framework: extpoint: excludes: – pro.shushi.pamirs.demo.core.extpoint.PetCatTypeExtPoint

    2024年5月13日
    2.0K00
  • IWrapper、QueryWrapper和LambdaQueryWrapper使用

    条件更新updateByWrapper 通常我们在更新的时候new一个对象出来在去更新,减少更新的字段 Integer update = new DemoUser().updateByWrapper(new DemoUser().setFirstLogin(Boolean.FALSE), Pops.<DemoUser>lambdaUpdate().from(DemoUser.MODEL_MODEL).eq(IdModel::getId, userId) 使用基础模型的updateById方法更新指定字段的方法: new 一下update对象出来,更新这个对象。 WorkflowUserTask userTaskUp = new WorkflowUserTask(); userTaskUp.setId(userTask.getId()); userTaskUp.setNodeContext(json); userTaskUp.updateById(); 条件删除updateByWrapper public List<T> delete(List<T> data) { List<Long> petTypeIdList = new ArrayList(); for(T item:data){ petTypeIdList.add(item.getId()); } Models.data().deleteByWrapper(Pops.<PetType>lambdaQuery().from(PetType.MODEL_MODEL).in(PetType::getId,petTypeIdList)); return data; } 构造条件查询数据 示例1: LambdaQueryWrapper拼接查询条件 private void queryPetShops() { LambdaQueryWrapper<PetShop> query = Pops.<PetShop>lambdaQuery(); query.from(PetShop.MODEL_MODEL); query.setSortable(Boolean.FALSE); query.orderBy(true, true, PetShop::getId); List<PetShop> petShops2 = new PetShop().queryList(query); System.out.printf(petShops2.size() + ""); } 示例2: IWrapper拼接查询条件 private void queryPetShops() { IWrapper<PetShop> wrapper = Pops.<PetShop>lambdaQuery() .from(PetShop.MODEL_MODEL).eq(PetShop::getId,1L); List<PetShop> petShops4 = new PetShop().queryList(wrapper); System.out.printf(petShops4.size() + ""); } 示例3: QueryWrapper拼接查询条件 private void queryPetShops() { //使用Lambda获取字段名,防止后面改字段名漏改 String nameField = LambdaUtil.fetchFieldName(PetTalent::getName); //使用Lambda获取Clumon名,防止后面改字段名漏改 String nameColumn = PStringUtils.fieldName2Column(nameField); QueryWrapper<PetShop> wrapper2 = new QueryWrapper<PetShop>().from(PetShop.MODEL_MODEL) .eq(nameColumn, "test"); List<PetShop> petShops5 = new PetShop().queryList(wrapper2); System.out.printf(petShops5.size() + ""); } IWrapper转为LambdaQueryWrapper @Function.Advanced(type= FunctionTypeEnum.QUERY) @Function.fun(FunctionConstants.queryPage) @Function(openLevel = {FunctionOpenEnum.API}) public Pagination<PetShopProxy> queryPage(Pagination<PetShopProxy> page, IWrapper<PetShopProxy> queryWrapper) { LambdaQueryWrapper<PetShopProxy> wrapper = ((QueryWrapper<PetShopProxy>) queryWrapper).lambda(); // 非存储字段从QueryData中获取 Map<String, Object> queryData = queryWrapper.getQueryData(); if (null != queryData && !queryData.isEmpty()) { String codes = (String) queryData.get("codes"); if (org.apache.commons.lang3.StringUtils.isNotBlank(codes)) { wrapper.in(PetShopProxy::getCode, codes.split(",")); } } return new PetShopProxy().queryPage(page, wrapper); }

    2024年5月25日
    2.9K00
  • GQL GET 请求生成图片文件流 — 技术实现方案

    GQL GET 请求生成图片文件流 — 技术实现方案 基于 Oinone/Pamirs GQL GET 文件流机制,参考 FileAction#downloadFormData(查询型)与 ExcelExportTaskAction#createExportTask(同步写流),实现「按业务 ID 动态生成图片并输出文件流」的通用指南。 1. 背景与目标 1.1 需求 在业务场景中,需要根据业务 ID(如订单 ID、模板 ID、单据 ID)在后端动态渲染图片(条码、标签、证书、预览图等),并将结果以文件流形式返回给前端。 1.2 约束 复用 Oinone 既有鉴权、会话、模块路由体系,不新增独立 REST 下载接口。 与 Excel 同步导出保持一致的调用方式:通过 GET 访问 /pamirs/{module}?query=…&variables=…。 前端仅两种消费方式: 页面预览:<img :src="imageUrl" /> 触发下载:window.open(imageUrl, '_blank') 2. 参考实现 2.1 查询型文件流(图片查看应对齐此模式) 图片查看属于只读查询,应使用 query + @Function(type = QUERY),与 CDN 文件下载一致: // @oinone/kunlun-vue-ui-common/…/UploadService.ts const gql = `query{resourceFileFormQuery{downloadFormData(resourceFileForm:{downloadUrl:"${url}"}){filename}}}`; window.open(UrlHelper.appendBasePath(`/pamirs/file?query=${encodeURIComponent(gql)}`), '_blank'); // pamirs-core/pamirs-file2/…/FileAction.java @Function.Advanced(displayName = "前端下载数据", type = FunctionTypeEnum.QUERY) public ResourceFileForm downloadFormData(ResourceFileForm resourceFileForm) { // 直接写 HttpServletResponse 输出流 response.setContentType("application/octet-stream"); response.setHeader("Content-Disposition", "attachment; filename=" + …); IOUtils.copy(inputStream, response.getOutputStream()); return new ResourceFileForm(); } 图片查看用 GQL.query,不用 GQL.mutation。 2.2 同步写流机制(共用) 无论 query 还是 mutation,文件流输出的底层机制相同:在 Function/Action 执行过程中直接写 HttpServletResponse。 2.2.1 HTTP 网关:RequestController GET /pamirs/{moduleName}?query={gql}&variables={json} POST /pamirs/{moduleName} body: { query, variables } GET 与 POST 走同一执行链路: // pamirs-framework/pamirs-gateways-graph-java/…/RequestController.java @RequestMapping(value = "/pamirs/{moduleName}", method = RequestMethod.GET) public DeferredResult<String> pamirsGet( @PathVariable("moduleName") String moduleName, @RequestParam("query") String gql, @RequestParam(value = "variables", required = false) String variables, …) { PamirsClientRequestParam gqlRequest = new PamirsClientRequestParam(); gqlRequest.setQuery(gql); if (!StringUtils.isBlank(variables)) { gqlRequest.setVariables(JsonUtils.parseMap(variables)); } return pamirsPost(moduleName, gqlRequest, request, response); } variables.path 在 RequestHelper.preparePamirsRequestParam 中经 SessionPrepareApi.prepare 注入会话上下文,与前端 getSessionPath() 对应。 2.2.2 Excel 导出对照:ExcelFileServiceImpl#doExportSync Excel 同步导出使用 mutation(createExportTask),因为会创建导出任务记录;图片查看无此副作用,故用 query。 同步导出的关键:劫持 FileClient.upload,将字节直接写入当前…

    2026年7月2日
    19900
  • Oinone开发实践-业务实现多租户方案

    总体方案 业务项目中,需要隔离的模型自定义增加租户字段进行数据隔离; 参考了Mybatis-Plus插件的TenantSqlParser进行的JPA实现,使用jsqlparser解析并修改SQL; 实现获取当前用户租户ID,SQL增删改查时处理租户字段,实现租户数据的隔离 参考项目: https://github.com/baomidou/mybatis-plus https://github.com/JSQLParser/JSqlParser 具体实现方式 1、业务上定义两个基础抽象模型包含租户字段 定义包含ID的基础抽象模型,且包含租户字段(如:公司编码, 用其他字段作为租户字段也可以,根据实际业务情况灵活修改)。 @Model.model(XXIdModel.MODEL_MODEL) @Model.Advanced(type = ModelTypeEnum.ABSTRACT) @Model(displayName = “带公司CODE的基础ID抽象模型”, summary = “带公司Code的Id模型”) public abstract class XXIdModel extends IdModel { public static final String MODEL_MODEL = “demo.biz.XXIdModel”; @Field.String @Field(displayName = “所属公司编码”, invisible = true, index = true) private String companyCode; } 定义包含Code的基础抽象模型,且包含租户字段(如:公司编码, 用其他字段作为租户字段也可以,根据实际业务情况灵活修改)。 @Model.model(XXCodeModel.MODEL_MODEL) @Model.Advanced(type = ModelTypeEnum.ABSTRACT) @Model(displayName = “带公司CODE的基础Code抽象模型”, summary = “带公司CODE的Code模型”) public abstract class XXCodeModel extends CodeModel { public static final String MODEL_MODEL = “demo.biz.XXCodeModel”; @Field.String @Field(displayName = “所属公司编码”, invisible = true, index = true) private String companyCode; } 2、业务模块的模型需租户隔离的都是继承上面这两个模型; @Model.model(PetPetCompany.MODEL_MODEL) @Model(displayName = “宠物公司”, labelFields = “name”) public class PetPetCompany extends AbstractCompanyCodeModel { public static final String MODEL_MODEL = “demo.PetPetCompany”; @Field.String @Field(displayName = “名称”) private String name; @Field.Text @Field(displayName = “简介”) private String introduction; } 3、自定义扩展Session,Session中设置租户信息 每次请求多把登录用户所属公司编码(companyCode)放到Session中;Session扩展参考:https://doc.oinone.top/oio4/9295.html 4、定义拦截器Interceptor进行数据隔离 数据创建和查询通过拦截器把Session中的中的公司编码(companyCode)设置到隔离字段中;拦截器的java示例代码参考: package pro.shushi.pamirs.demo.core.interceptor; import net.sf.jsqlparser.JSQLParserException; import net.sf.jsqlparser.expression.Expression; import net.sf.jsqlparser.expression.LongValue; import net.sf.jsqlparser.expression.StringValue; import net.sf.jsqlparser.expression.operators.relational.ExpressionList; import net.sf.jsqlparser.expression.operators.relational.ItemsListVisitor; import net.sf.jsqlparser.expression.operators.relational.MultiExpressionList; import net.sf.jsqlparser.expression.operators.relational.NamedExpressionList; import net.sf.jsqlparser.parser.CCJSqlParserUtil; import net.sf.jsqlparser.schema.Column; import net.sf.jsqlparser.statement.Statement; import net.sf.jsqlparser.statement.insert.Insert; import net.sf.jsqlparser.statement.select.*; import net.sf.jsqlparser.statement.update.Update; import net.sf.jsqlparser.statement.values.ValuesStatement; import org.apache.commons.lang3.StringUtils; import org.apache.ibatis.executor.Executor; import org.apache.ibatis.mapping.BoundSql; import org.apache.ibatis.mapping.MappedStatement; import org.apache.ibatis.mapping.SqlCommandType; import org.apache.ibatis.mapping.SqlSource; import org.apache.ibatis.plugin.*; import org.apache.ibatis.reflection.MetaObject; import org.apache.ibatis.reflection.SystemMetaObject; import…

    2024年4月6日
    1.4K00
  • Schedule相关

    1、Schedule初始化 TODO 2、Schedule执行器的入口 通常本地创建了Schedule,没有被正常执行,可以通过这个入口去排查问题 pro.shushi.pamirs.middleware.schedule.core.tasks.AbstractScheduleTaskDealSingle#selectTasks 3、Schedule执行环境隔离 项目中开发如果本地进行任务调试,通过通过指定ownSign进行环境隔离,如果不配置可能会导致这个任务被别的机器执行,本机的代码无法调试,如果开发的时候出现任务未执行可能是这个原因导致的 event: enabled: true schedule: enabled: true ownSign: dev_wx rocket-mq: namesrv-addr: 127.0.0.1:9876

    后端 2023年11月16日
    2.2K00

Leave a Reply

Please Login to Comment