前言

上一节《4.snail-job分片任务》中已经对分片任务有了大致的了解。只不过上一节讲述的是静态分片任务，在本节中要介绍的Map任务是-动态分片任务。所谓的"静态"和"动态",其本质区别在于是开始就配置好了分片还是由代码逻辑来分片。直接配置的就是“静态”；而代码分配的就是“动态“。

不管是静态分片，还是动态分片，其核心原理是将一个大任务拆分成多个小任务并发执行。本节介绍的Map任务和下一节将要介绍的MapReduce任务都属于动态分片范畴。

本节目标

这里用一个比较简单的示例，来演示Map任务。我们有200个数字，切成4个片，每个片中有50个数。子任务就是要对分配给自己的50个数字进行相加计算。OK，就这么简单。但是依然能从实现的过程中了解如下知识点：

• 客户端采用继承类方式实现上述计算功能
• 客户端采用注解方式实现上述计算功能
• 服务端如何配置Map动态分片任务
• 服务端查看分片任务的执行情况
• 服务端配置并行数提高性能

客户端代码

开发环境

• JDK版本：openjdk-21.0.2
• snail-job版本：1.2.0

Maven依赖

xml
 代码解读
复制代码
<dependencies>
        <dependency>
            <groupId>org.springframework.bootgroupId>
            <artifactId>spring-boot-starter-webartifactId>
        dependency>
        
        <dependency>
            <groupId>com.aizudagroupId>
            <artifactId>snail-job-client-starterartifactId>
            <version>1.2.0version>
        dependency>
        
        <dependency>
            <groupId>com.aizudagroupId>
            <artifactId>snail-job-client-retry-coreartifactId>
            <version>1.2.0version>
        dependency>
        
        <dependency>
            <groupId>com.aizudagroupId>
            <artifactId>snail-job-client-job-coreartifactId>
            <version>1.2.0version>
        dependency>
    dependencies>

继承类方式

scala
 代码解读
复制代码
@Component
public class TestMapJob extends AbstractMapExecutor {
    @Override
    public ExecuteResult doJobMapExecute(MapArgs mapArgs, MapHandler mapHandler) {
        return switch (mapArgs.getTaskName()) {
            case SystemConstants.ROOT_MAP -> {
                // 生成1~200数值并分片
                int partitionSize = 50;
                List<List<Integer>> partition = IntStream.rangeClosed(1, 200)
                        .boxed()
                        .collect(Collectors.groupingBy(i -> (i - 1) / partitionSize))
                        .values()
                        .stream()
                        .toList();
                SnailJobLog.REMOTE.info("端口:{}完成分配任务", SpringUtil.getProperty("server.port"));
                yield mapHandler.doMap(partition, "doCalc");
            }
            case "doCalc" -> {
                List<Integer> sourceList = (List<Integer>) mapArgs.getMapResult();
                // 遍历sourceList的每一个元素,计算出一个累加值partitionTotal
                int partitionTotal = sourceList.stream().mapToInt(i -> i).sum();
                // 打印日志到服务器
                ThreadUtil.sleep(3, TimeUnit.SECONDS);
                SnailJobLog.REMOTE.info("端口:{},partitionTotal:{}", SpringUtil.getProperty("server.port"), partitionTotal);
                yield ExecuteResult.success(partitionTotal);
            }
            default -> ExecuteResult.failure();
        };
    }
}

解释说明：

• 通过继承AbstractMapExecutor类来实现动态分片
• 通过判断任务名，来处理到底是应该分片还是对分片后的处理。在snail-job框架中，定义分片任务名是ROOT_MAP。并且仅由一台机器执行。假设在集群中有4台机器，那么这个ROOT_MAP任务，会随机从集群中抽取一个存活的机器进行执行。
• 非ROOT_MAP名的任务，就是分片后的处理。
• 需要对分片后的结果进行处理用mapHandler.doMap(partition, "doCalc")；不需要处理就return ExecuteResult`。

注解方式

scss
 代码解读
复制代码
@Component
@JobExecutor(name = "testMapJobAnnotation")
public class TestMapJobAnnotation {
​
    @MapExecutor
    public ExecuteResult doJobMapExecute(MapArgs mapArgs, MapHandler mapHandler) {
        // 生成1~200数值并分片
        int partitionSize = 50;
        List> partition = IntStream.rangeClosed(1, 200)
                .boxed()
                .collect(Collectors.groupingBy(i -> (i - 1) / partitionSize))
                .values()
                .stream()
                .toList();
        SnailJobLog.REMOTE.info("端口:{}完成分配任务", SpringUtil.getProperty("server.port"));
        return mapHandler.doMap(partition, "doCalc");
    }
​
    @MapExecutor(taskName = "doCalc")
    public ExecuteResult doCalc(MapArgs mapArgs) {
        List sourceList = (List) mapArgs.getMapResult();
        // 遍历sourceList的每一个元素,计算出一个累加值partitionTotal
        int partitionTotal = sourceList.stream().mapToInt(i -> i).sum();
        // 打印日志到服务器
        ThreadUtil.sleep(3, TimeUnit.SECONDS);
        SnailJobLog.REMOTE.info("端口:{},partitionTotal:{}", SpringUtil.getProperty("server.port"), partitionTotal);
        return ExecuteResult.success(partitionTotal);
    }
}

解释说明：

• 通过@MapExecutor标识要执行的任务名称。这个注解默认的任务名称就是ROOT_MAP

  less
   代码解读
  复制代码
  @Target({ElementType.METHOD})
  @Retention(RetentionPolicy.RUNTIME)
  @Inherited
  @Documented
  public @interface MapExecutor {
  ​
      /**
       * 任务名称
       *
       * @return
       */
      String taskName() default SystemConstants.ROOT_MAP;
  ​
  }
  

• 其实用法和继承类的方式是一样的。都通过任务名称来执行具体逻辑。前一个任务处理的结果，后个任务通过MapArgs参数获得，继续处理自己的逻辑。甚至还可以再把自己处理的结果，交给下一个任务名来处理。

服务端配置

继承类方式

说明：

• 状态：状态设置为禁用，是想通过手动执行来触发
• 任务类型：要选本节介绍的任务类型Map
• 自定义执行器：继承类的方式要写全路径【复习】
• 并行数：指客户端每台机器的线程数【复习】

注解方式

说明：

• 自定义执行器 ： 与注解@JobExecutor(name = "testMapJobAnnotation")名称一致【复习】

进行测试

测试前提

这里测试采用2台客户端。

可以参考《3.snail-job广播任务》的本机两个客户端启动章节的介绍。idea中配置如下：

ini
 代码解读
复制代码
-Dserver.port=9100 -Dsnail-job.port=1900
-Dserver.port=9200 -Dsnail-job.port=2900

测试MAP任务

• 9100Web端口

  

• 9200Web端口

  

服务端管理页面

• 查看Map的执行批次信息

  

• 展开ROOT_MAP

  

• 还可以查看结果

  点击上截图页面中结果列中的查看结果按钮

  

• 再看看并行数修改为2后执行效率的提升

  

总结

• 静态分片就是通过配置参数定义每个分片的范围；动态分配就是程序逻辑决定如何分片。
• 通过程序代码的使用体验来看，注解方式似乎更直观些。
• ROOT_MAP任务时框架定义的用于进行分片的根任务。
• ROOT_MAP任务，会随机抽取集群中一台存活机器进行执行。
• Map任务的核心是通过任务名称来进行的。分片动作的执行由ROOT_MAP指定，分片后的任务由你指定。