在Spring中实现Bulk和Batch API

最后修改:2024年7月17日

by Saikat Chakraborty

1. 概述

标准REST API能覆盖大部分典型场景，但在处理批量操作时存在局限性。本文将探讨如何在微服务中实现Bulk和Batch操作，并展示几种自定义的写操作批量API实现方案。

2. Bulk和Batch API简介

⚠️ 术语澄清：虽然Bulk和Batch常被混用，但两者有本质区别：

  • Bulk操作：对同类型资源执行相同操作（如批量创建用户）
  • Batch操作：对多种资源执行不同操作（如同时创建用户和更新订单）

✅ 核心优势：

  • 减少网络往返次数
  • 降低整体延迟
  • 提升应用性能

⚠️ 踩坑点：

  • 主流框架（如Spring）没有内置支持
  • 缺乏统一标准规范
  • 需要自定义实现

3. Spring示例应用

3.1 Maven依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
    <version>3.1.5</version>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-validation</artifactId>
    <version>3.1.5</version>
</dependency>

3.2 实现第一个Spring服务

Customer模型：

public class Customer {
    private int id;
    private String name;
    private String email;
    private String address;
    // 标准getter/setter
}

CustomerService实现：

@Service
public class CustomerService {
    private final Map<String, Customer> customerRepoMap = new HashMap<>();

    public List<Customer> createCustomers(List<Customer> customers) {
        return customers.stream()
          .map(this::createCustomer)
          .filter(Optional::isPresent)
          .map(Optional::get)
          .collect(toList());
    }

    public Optional<Customer> createCustomer(Customer customer) {
        if (!customerRepoMap.containsKey(customer.getEmail()) && customer.getId() == 0) {
            Customer customerToCreate = new Customer(customerRepoMap.size() + 1, 
              customer.getName(), customer.getEmail());
            customerToCreate.setAddress(customer.getAddress());
            customerRepoMap.put(customerToCreate.getEmail(), customerToCreate);  
            return Optional.of(customerToCreate);
        }
        return Optional.empty();
    }

    private Optional<Customer> updateCustomer(Customer customer) {
        Customer customerToUpdate = customerRepoMap.get(customer.getEmail());
        if (customerToUpdate != null && customerToUpdate.getId() == customer.getId()) {
            customerToUpdate.setName(customer.getName());
            customerToUpdate.setAddress(customer.getAddress());
        }
        return Optional.ofNullable(customerToUpdate);
    }

    public Optional<Customer> deleteCustomer(Customer customer) {
        Customer customerToDelete = customerRepoMap.get(customer.getEmail());
        if (customerToDelete != null && customerToDelete.getId() == customer.getId()) {
            customerRepoMap.remove(customer.getEmail());
        }
        return Optional.ofNullable(customerToDelete);
    }
}

3.3 实现第二个Spring服务

Address模型：

public class Address implements Serializable {
    private int id;
    private String street;
    private String city;
    // 标准getter/setter
}

AddressService实现：

@Service
public class AddressService {
    private final Map<String, Address> addressRepoMap = new HashMap<>();

    public Address createAddress(Address address) {
        Address createdAddress = null;
        String addressUniqueKey = address.getStreet().concat(address.getCity());
        if (!addressRepoMap.containsKey(addressUniqueKey)) {
            createdAddress = new Address(addressRepoMap.size() + 1, 
              address.getStreet(), address.getCity());
            addressRepoMap.put(addressUniqueKey, createdAddress);
        }
        return createdAddress;
    }
}

4. 基于现有接口实现Bulk API

4.1 实现Bulk控制器

请求格式：

[
    {
        "name": "张三",
        "email": "zhangsan@example.com",
        "address": "北京市朝阳区"
    }
]

控制器实现：

@RestController
@RequestMapping("/api")
public class BulkController {
    private final CustomerService customerService;

    public BulkController(CustomerService customerService) {
        this.customerService = customerService;
    }

    @PostMapping(path = "/customers/bulk")
    public ResponseEntity<List<Customer>> bulkCreateCustomers(
      @RequestHeader(value="X-ActionType") String actionType, 
      @RequestBody @Valid @Size(min = 1, max = 20) List<Customer> customers) {
        List<Customer> customerList = actionType.equals("bulk") ? 
          customerService.createCustomers(customers) :
          singletonList(customerService.createCustomer(customers.get(0)).orElse(null));

        return new ResponseEntity<>(customerList, HttpStatus.CREATED);
    }
}

4.2 验证Bulk API

测试请求：

curl -i --request POST 'http://localhost:8080/api/customers/bulk' \
--header 'X-ActionType: bulk' \
--header 'Content-Type: application/json' \
--data-raw '[
    {
        "name": "李四",
        "email": "lisi@example.com",
        "address": "上海市浦东新区"
    },
    {
        "name": "王五",
        "email": "wangwu@example.com",
        "address": "广州市天河区"
    }
]'

响应结果：

HTTP/1.1 201 
[{"id":1,"name":"李四","email":"lisi@example.com","address":"上海市浦东新区"},
{"id":2,"name":"王五","email":"wangwu@example.com","address":"广州市天河区"}]

5. 使用独立接口实现Bulk API

5.1 定义请求和响应模型

请求格式：

[
    {
        "bulkActionType": "CREATE",
        "customers": [
            {
                "name": "赵六",
                "email": "zhaoliu@example.com",
                "address": "深圳市南山区"
            }
        ]
    }
]

模型类：

public class CustomerBulkRequest {
    private BulkActionType bulkActionType;
    private List<Customer> customers;
    // 标准getter/setter
}

public enum BulkActionType {
    CREATE, UPDATE, DELETE
}

public class CustomerBulkResponse {
    private BulkActionType bulkActionType;
    private List<Customer> customers;
    private BulkStatus status;
    // 标准getter/setter
}

public enum BulkStatus {
    PROCESSED, PARTIALLY_PROCESSED, NOT_PROCESSED
}

5.2 实现Bulk控制器

@RestController
@RequestMapping("/api")
@Validated
public class BulkController {
    private final CustomerService customerService;
    private final EnumMap<BulkActionType, Function<Customer, Optional<Customer>>> bulkActionFuncMap = 
      new EnumMap<>(BulkActionType.class);

    public BulkController(CustomerService customerService) {
        this.customerService = customerService;
        bulkActionFuncMap.put(BulkActionType.CREATE, customerService::createCustomer);
        bulkActionFuncMap.put(BulkActionType.UPDATE, customerService::updateCustomer);
        bulkActionFuncMap.put(BulkActionType.DELETE, customerService::deleteCustomer);
    }

    @PostMapping(path = "/customers/bulk")
    public ResponseEntity<List<CustomerBulkResponse>> bulkProcessCustomers(
      @RequestBody @Valid @Size(min = 1, max = 20) 
      List<CustomerBulkRequest> customerBulkRequests) {
        List<CustomerBulkResponse> customerBulkResponseList = new ArrayList<>();

        customerBulkRequests.forEach(customerBulkRequest -> {
            List<Customer> customers = customerBulkRequest.getCustomers().stream()
              .map(bulkActionFuncMap.get(customerBulkRequest.getBulkActionType()))
              .filter(Optional::isPresent)
              .map(Optional::get)
              .collect(toList());
            
            BulkStatus bulkStatus = getBulkStatus(customerBulkRequest.getCustomers(), 
              customers);     
            customerBulkResponseList.add(CustomerBulkResponse.getCustomerBulkResponse(customers, 
              customerBulkRequest.getBulkActionType(), bulkStatus));
        });

        return new ResponseEntity<>(customerBulkResponseList, HttpStatus.MULTI_STATUS);
    }

    private BulkStatus getBulkStatus(List<Customer> customersInRequest, 
      List<Customer> customersProcessed) {
        if (!customersProcessed.isEmpty()) {
            return customersProcessed.size() == customersInRequest.size() ?
              BulkStatus.PROCESSED : 
              BulkStatus.PARTIALLY_PROCESSED;
        }
        return BulkStatus.NOT_PROCESSED;
    }
}

5.3 验证Bulk API

测试请求：

curl -i --request POST 'http://localhost:8080/api/customers/bulk' \
--header 'Content-Type: application/json' \
--data-raw '[
    {
        "bulkActionType": "CREATE",
        "customers": [
            {
                "name": "钱七",
                "email": "qianqi@example.com",
                "address": "杭州市西湖区"
            }
        ]
    },
    {
        "bulkActionType": "UPDATE",
        "customers": [
            {
                "id": 1,
                "name": "李四更新",
                "email": "lisi@example.com",
                "address": "北京市海淀区"
            }
        ]
    }
]'

响应结果：

HTTP/1.1 207 
[{"customers":[{"id":3,"name":"钱七","email":"qianqi@example.com","address":"杭州市西湖区"}],"status":"PROCESSED","bulkType":"CREATE"},
{"customers":[{"id":1,"name":"李四更新","email":"lisi@example.com","address":"北京市海淀区"}],"status":"PROCESSED","bulkType":"UPDATE"}]

6. 实现Batch API

6.1 实现Batch请求模型

请求格式：

[
    {
        "method": "POST",
        "relativeUrl": "/address",
        "data": {
            "street": "科技园路",
            "city": "南京市"
        }
    },
    {
        "method": "PATCH",
        "relativeUrl": "/customer",
        "data": {
            "id": 2,
            "name": "王五更新",
            "email": "wangwu@example.com",
            "address": "成都市高新区"
        }
    }
]

模型类：

public class BatchRequest {
    private HttpMethod method;
    private String relativeUrl;
    private JsonNode data;
    // 标准getter/setter
}

6.2 实现Batch控制器

@RestController
@RequestMapping("/api")
public class BatchController {
    private final CustomerService customerService;
    private final AddressService addressService;
    private final ObjectMapper objectMapper;

    public BatchController(CustomerService customerService, 
                         AddressService addressService,
                         ObjectMapper objectMapper) {
        this.customerService = customerService;
        this.addressService = addressService;
        this.objectMapper = objectMapper;
    }

    @PostMapping(path = "/batch")
    public String batchUpdateCustomerWithAddress(
      @RequestBody @Valid @Size(min = 1, max = 20) List<BatchRequest> batchRequests) {
        batchRequests.forEach(batchRequest -> {
            if (batchRequest.getMethod().equals(HttpMethod.POST) && 
              batchRequest.getRelativeUrl().equals("/address")) {
                addressService.createAddress(objectMapper.convertValue(batchRequest.getData(), 
                  Address.class));
            } else if (batchRequest.getMethod().equals(HttpMethod.PATCH) && 
                batchRequest.getRelativeUrl().equals("/customer")) {
                customerService.updateCustomer(objectMapper.convertValue(batchRequest.getData(), 
                  Customer.class));
            }
        });
        return "Batch update is processed";
    }
}

6.3 验证Batch API

测试请求：

curl -i --request POST 'http://localhost:8080/api/batch' \
--header 'Content-Type: application/json' \
--data-raw '[
    {
        "method": "POST",
        "relativeUrl": "/address",
        "data": {
            "street": "天府大道",
            "city": "成都市"
        }
    },
    {
        "method": "PATCH",
        "relativeUrl": "/customer",
        "data": {
            "id": 2,
            "name": "王五更新",
            "email": "wangwu@example.com",
            "address": "成都市高新区"
        }
    }
]'

响应结果：

HTTP/1.1 200
Batch update is processed

7. 总结

本文深入探讨了在Spring应用中实现Bulk和Batch操作的几种方案：

✅ 关键要点：

  1. Bulk操作：对同类型资源执行相同操作

    • 方案1：复用现有接口（通过请求头区分）
    • 方案2：独立接口（支持多种操作类型）

  2. Batch操作：对不同资源执行不同操作

    • 通过HttpMethod和relativeUrl路由请求
    • 使用ObjectMapper处理动态数据类型

⚠️ 最佳实践：

  • 添加请求大小限制（@Size注解）
  • 实现部分成功处理机制
  • 考虑事务边界（原子性 vs 非原子性）
  • 添加操作冲突校验

完整示例代码可在GitHub仓库获取。

原始标题:Implement Bulk and Batch API in Spring | Baeldung

« 上一篇： Apache Kafka Series | Baeldung
» 下一篇： 使用 Jackson 序列化和反序列化 java.sql.Blob