【开源项目】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 {

  @Override
  public String[] selectImports(AnnotationMetadata metadata) {
    return new String[] {
      StreamScannerRegistrar.class.getName(), StreamPluginAutoConfiguration.class.getName()
    };
  }

  @Override
  public int getOrder() {
    return HIGHEST_PRECEDENCE;
  }
}

StreamScannerRegistrar注入了StreamScannerConfigurer扫描类。

public class StreamScannerRegistrar implements ImportBeanDefinitionRegistrar {

  @Override
  public 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。

  @Override
  public 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来完成异步执行。

    @SafeVarargs
    public 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();
    }