【开源项目】Stream-Query的入门使用和原理分析

前言

无意间发现了一个有趣的项目，Stream-Query。了解了一下其基本的功能，可以帮助开发者省去Mapper的编写。在开发中，我们会编写entity和mapper来完成业务代码，但是Stream-Query可以省去mapper，只写entity。

快速入门

实体类

@Data
public class UserInfo{private static final long serialVersionUID = -7219188882388819210L;@TableId(value = "id", type = IdType.AUTO)private Long id;private String name;private Integer age;private String email;
}

创表语句

create table user_info
(id   bigint auto_increment comment '主键'primary key,name varchar(20) null comment '姓名',age  int comment '年龄',email  varchar(20)  comment '邮件' 
)comment '用户信息';

配置扫描包

@EnableMybatisPlusPlugin(basePackages = "com.charles.entity.**")

插入Demo

    @GetMapping("/t1")public void t1() {UserInfo userInfo = new UserInfo();userInfo.setAge(12);userInfo.setEmail("123@qq.com");userInfo.setName("张三");UserInfo userInfo2 = new UserInfo();userInfo2.setAge(123);userInfo2.setEmail("123@qq.com");userInfo2.setName("李四");Database.saveFewSql(Arrays.asList(userInfo, userInfo2));}

单个查询Demo

    @GetMapping("t2")public void t2() {UserInfo userInfo = One.of(UserInfo::getId).eq(2L).query();System.out.println(userInfo);}

多个查询Demo

    @GetMapping("t3")public void t3() {QueryCondition.query(UserInfo.class).in(UserInfo::getName, Lists.of("张三", "李四"));QueryCondition<UserInfo> wrapper =QueryCondition.query(UserInfo.class).in(UserInfo::getName, Lists.of("张三", "李四"));List<UserInfo> list = Database.list(wrapper);Map<Long, UserInfo> idUserMap = OneToOne.of(UserInfo::getId).eq(1L).query();System.out.println(list);}

Stream-Query通过Database，One，Many等静态方法完成查询和插入等操作。

核心原理分析

EnableMybatisPlusPlugin注入了StreamConfigurationSelector。

@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.TYPE})
@Documented
@Inherited
@Import({StreamConfigurationSelector.class})
public @interface EnableMybatisPlusPlugin {/*** Alias for {@link #basePackages()}** @return base packages*/String[] value() default {};/*** Base packages** @return base packages*/String[] basePackages() default {};
}

StreamConfigurationSelector注入了StreamScannerRegistrar扫描注册器和StreamPluginAutoConfiguration配置类。

public class StreamConfigurationSelector implements DeferredImportSelector, Ordered {@Overridepublic String[] selectImports(AnnotationMetadata metadata) {return new String[] {StreamScannerRegistrar.class.getName(), StreamPluginAutoConfiguration.class.getName()};}@Overridepublic int getOrder() {return HIGHEST_PRECEDENCE;}
}

StreamScannerRegistrar注入了StreamScannerConfigurer扫描类。

public class StreamScannerRegistrar implements ImportBeanDefinitionRegistrar {@Overridepublic void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) {AnnotationAttributes annotationAttributes =AnnotationAttributes.fromMap(importingClassMetadata.getAnnotationAttributes(EnableMybatisPlusPlugin.class.getName()));if (Objects.isNull(annotationAttributes)) {return;}BeanDefinitionBuilder builder =BeanDefinitionBuilder.genericBeanDefinition(StreamScannerConfigurer.class);Set<String> basePackages = new HashSet<>();basePackages.addAll(Arrays.stream(annotationAttributes.getStringArray("value")).filter(StringUtils::hasText).collect(Collectors.toSet()));basePackages.addAll(Arrays.stream(annotationAttributes.getStringArray("basePackages")).filter(StringUtils::hasText).collect(Collectors.toSet()));basePackages.addAll(Arrays.stream(annotationAttributes.getClassArray("basePackageClasses")).filter(Objects::nonNull).map(ClassUtils::getPackageName).collect(Collectors.toSet()));if (basePackages.isEmpty()) {basePackages.add(ClassUtils.getPackageName(importingClassMetadata.getClassName()));builder.addPropertyValue("emptyBasePackages", true);}builder.addPropertyValue("basePackages", basePackages);Set<Class<?>> classes =Arrays.stream(annotationAttributes.getClassArray("classes")).filter(Objects::nonNull).collect(Collectors.toSet());builder.addPropertyValue("classes", classes);Class<? extends Annotation> annotation = annotationAttributes.getClass("annotation");if (!Annotation.class.equals(annotation)) {builder.addPropertyValue("annotation", annotation);}Class<?> scanInterface = annotationAttributes.getClass("interfaceClass");if (!Class.class.equals(scanInterface)) {builder.addPropertyValue("interfaceClass", scanInterface);}registry.registerBeanDefinition("streamScannerConfigurer", builder.getBeanDefinition());}
}

StreamScannerConfigurer实现了BeanFactoryPostProcessor，StreamScannerConfigurer#postProcessBeanFactory可以根据注解扫描，可以根据接口扫描，可以根据扫描包扫描。详情可见 enablemybatisplusplugin。

  @Overridepublic void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory)throws BeansException {defaultScanConfig();// 指定类registerEntityClasses(this.classes);StreamClassPathScanner scanner = new StreamClassPathScanner(false);scanner.setAnnotation(this.annotation);scanner.setInterfaceClass(this.interfaceClass);scanner.registerFilters();Set<Class<?>> classSet = scanner.scan(this.basePackages);registerEntityClasses(classSet);}

在StreamPluginAutoConfiguration配置类中注入了DynamicMapperHandler。

  @Bean@ConditionalOnMissingBean(DynamicMapperHandler.class)public DynamicMapperHandler dynamicMapperHandler(SqlSessionFactory sqlSessionFactory, StreamScannerConfigurer streamScannerConfigurer) {return new DynamicMapperHandler(sqlSessionFactory, streamScannerConfigurer.getEntityClasses());}

DynamicMapperHandler该类的作用就是根据传入的entity的列表，构建Mapper。

public class DynamicMapperHandler {public DynamicMapperHandler(SqlSessionFactory sqlSessionFactory, Collection<Class<?>> entityClassList) {Configuration configuration = sqlSessionFactory.getConfiguration();if (configuration instanceof MybatisConfiguration) {MybatisConfiguration mybatisConfiguration = (MybatisConfiguration) configuration;entityClassList.forEach(entityClass -> Database.buildMapper(mybatisConfiguration, entityClass));}}
}

Database#buildMapper，根据ByteBuddy来生成对应的接口实现。

  public static void buildMapper(Configuration configuration, Class<?> entityClass) {if (!(configuration instanceof MybatisConfiguration)) {throw new IllegalArgumentException("configuration must be MybatisConfiguration");}Maps.computeIfAbsent(ENTITY_MAPPER_CLASS_CACHE,entityClass,k -> {Class<?> dynamicMapper =new ByteBuddy().makeInterface(TypeDescription.Generic.Builder.parameterizedType(IMapper.class, entityClass).build()).name(String.format("%s.%sMapper",PluginConst.DYNAMIC_MAPPER_PREFIX, entityClass.getSimpleName())).make().load(ClassUtils.class.getClassLoader()).getLoaded();configuration.addMapper(dynamicMapper);return dynamicMapper;});}

以上就是项目初始化的流程，StreamQuery帮助我们完成了根据Entity来自动生成Mapper，接下来我们分析一下StreamQuery是如何帮助我们简化使用的。

Database#saveFewSql(java.util.Collection<T>)，保存操作，获取SqlSession，获取IMapper，执行saveFewSql的方法。

  public static <T> boolean saveFewSql(Collection<T> entityList) {return saveFewSql(entityList, PluginConst.DEFAULT_BATCH_SIZE);}public static <T> boolean saveFewSql(Collection<T> entityList, int batchSize) {if (CollectionUtils.isEmpty(entityList) || batchSize <= 0) {return false;}return execute(getEntityClass(entityList),(IMapper<T> baseMapper) ->entityList.size() == baseMapper.saveFewSql(entityList, batchSize));}public static <T, R, M extends BaseMapper<T>> R execute(Class<T> entityClass, SFunction<M, R> sFunction) {SqlSession sqlSession = SqlHelper.sqlSession(entityClass);try {return sFunction.apply(getMapper(entityClass, sqlSession));} finally {SqlSessionUtils.closeSqlSession(sqlSession, GlobalConfigUtils.currentSessionFactory(entityClass));}}

IMapper#saveFewSql，默认实现是批量拆分List，调用saveOneSql。

  /*** 批量插入** @param list 集合* @param batchSize 分割量* @return 是否成功*/default long saveFewSql(Collection<T> list, int batchSize) {return Steam.of(list).splitList(batchSize).mapToLong(this::saveOneSql).sum();}

补充了解

`One`

One，返回单个实体类。通过封装Database来完成查询单个操作。

  /*** query.** @return a V object*/public V query() {return Sf.of(Database.getOne(wrapper)).mayAlso(peekConsumer).mayLet(valueOrIdentity()).get();}

`QueryCondition`

QueryCondition查询条件类，继承了LambdaQueryWrapper。

也就是new LambdaQueryWrapper<UserInfo>().in(UserInfo::getName, Lists.of("张三", "李四"));等于QueryCondition<UserInfo> wrapper = QueryCondition.query(UserInfo.class).in(UserInfo::getName, Lists.of("张三", "李四"));

public class QueryCondition<T> extends LambdaQueryWrapper<T> {public static <T> QueryCondition<T> query(Class<T> entityClass) {QueryCondition<T> condition = new QueryCondition<>();condition.setEntityClass(entityClass);return condition;}

`OneToOne`

OneToOne封装了一层Stream的操作。

  public Map<K, V> query() {return query(HashMap::new);}/*** query.** @param mapFactory a {@link java.util.function.IntFunction} object* @param <R> a R class* @return a R object*/public <R extends Map<K, V>> R query(IntFunction<R> mapFactory) {List<T> list = Database.list(wrapper);return Steam.of(list).parallel(isParallel).peek(peekConsumer).toMap(keyFunction,valueOrIdentity(),SerBiOp.justAfter(),() -> mapFactory.apply(list.size()));}

`AsyncHelper`

AsyncHelper使用

    public static void main(String[] args) {List<String> result = AsyncHelper.supply(() -> {System.out.println(Thread.currentThread().getName() + "1111");return "123";}, () -> {System.out.println(Thread.currentThread().getName() + "2345");return "456";});System.out.println(result);}

原理分析，可以指定拦截器和线程池，使用CompletableFuture.supplyAsync来完成异步执行。

    @SafeVarargspublic static <T> List<T> supply(AsyncConfig asyncConfig, SerSupp<T>... suppliers) {AsyncInterceptor interceptor = asyncConfig.getInterceptor();interceptor.before();CompletableFuture<T>[] futures = (CompletableFuture[])Steam.of(suppliers).map((supplier) -> {return CompletableFuture.supplyAsync(() -> {return interceptor.execute(supplier);}, asyncConfig.getExecutor());}).toArray((x$0) -> {return new CompletableFuture[x$0];});CompletableFuture var10000 = CompletableFuture.allOf(futures);interceptor.getClass();CompletableFuture<Void> exceptionally = var10000.exceptionally(interceptor::onError);(() -> {return asyncConfig.getTimeout() == -1 ? exceptionally.get() : exceptionally.get((long)asyncConfig.getTimeout(), asyncConfig.getTimeUnit());}).get();interceptor.after();return Steam.of(futures).map(CompletableFuture::get).toList();}