Scalable PBC Architecture for Decoupled Logistics Workflow Orchestration

Problem

Traditional logistics services hard‑code each business scenario in controllers, resulting in deep if‑else nesting, high coupling between steps, inability to reuse a step across flows, and business rules scattered throughout the code base.

Spaghetti code : nested if‑else in controllers.

High coupling : adding or changing a step requires modifications in many places.

No reuse : the same step cannot be shared.

Configuration entangled with code : business rules are stored in code.

Solution – Process Business Capability (PBC) Architecture

Each logistics step is extracted as an independent PBC node annotated with @BusinessObject. A flow engine matches a request to a flow definition stored in a database and executes the nodes sequentially.

Step decoupling : a node only contains its own logic; data is passed explicitly via FlowContext.

Configurable orchestration : process definitions and node order are stored as JSON in tb_flow and can be changed without code changes.

Unified entry point : /api/pbc/flow/execute routes all scenarios.

External rule storage : node‑level constraints and switches are kept in tb_nodes as JSON.

Core Concepts

PBC

: reusable step node (e.g., Transfer, Sorting, Shipping). BusinessObject: annotation that declares a class as a PBC node and defines module, nodeCode, nodeName. Flow: ordered list of node codes with data‑mapping definitions and matching conditions. FlowEngine: core scheduler that performs flow matching, chained execution, and unified persistence. FlowContext: carries the original request payload and a map of node outputs. ExternalRestrictions: AOP‑based node‑level validation before execution. Configuration: node‑internal switches that affect only the node’s internal logic.

Node Structure (example: Vehicle Arrival)

Identity :

@BusinessObject(module="Car", nodeCode="CAR_ARR_001", nodeName="Vehicle Arrival")

External restriction : @ExternalRestrictions on the execute method.

Configuration : JSON stored in tb_nodes.configuration (e.g., {"is_support_receipt":true}).

Input/Output : receives FlowContext, reads upstream data, writes a NodeResult back to the context.

Architecture Layers

Unified entry layer : FlowController receives requests and delegates to the engine.

Flow engine layer : FlowMatcher (matching) + FlowExecutor (execution).

AOP layer : ExternalRestrictions validates node‑level constraints.

PBC node layer : classes annotated with @BusinessObject implement PbcNode.

Configuration layer : JSON configuration per node.

Data layer : MySQL tables tb_flow, tb_nodes, tb_flow_log.

Metadata management : PbcMetadataScanner scans @BusinessObject at startup and registers them in BusinessObjectRegistry, which is exposed via /api/pbc/metadata/nodes.

Before‑After Comparison

Process definition : hard‑coded if‑else vs. configurable JSON in tb_flow.

Adding a scenario : new controller & service vs. insert a new flow record.

Changing step order : modify many code locations vs. update tb_flow.nodes order.

Step reuse : duplicate development vs. reuse the same PBC node across flows.

Business rule change : code change & redeploy vs. update database configuration.

Observability : scattered interfaces vs. unified entry + execution logs in tb_flow_log.

Flow Engine Implementation

Flow Matching ( FlowMatcher )

All active flows are loaded from tb_flow. The request payload is converted to a JSON tree and each flow’s conditions JSON is evaluated:

All rules must satisfy the logical operator (AND/OR).

If no flow matches, a runtime exception "未找到匹配的流程" is thrown.

If multiple flows match, a conflict exception is thrown with both process names.

private boolean evaluateConditions(NodeRestrictions restrictions, JsonNode requestJson) { ... }

private boolean evaluateRule(NodeRestrictions.Rule rule, String actualValue) { ... }

Flow Execution ( FlowExecutor )

Execution steps:

Create a FlowContext and initialise it with the request payload.

Iterate over FlowDefinition.NodeStep in order:

ExternalRestrictions AOP validates node‑level constraints; failure triggers fast‑fail.

Resolve the node from the Spring context ( resolveNode).

Apply data‑mapping rules ( applyMappings).

Invoke node.execute(context) and record execution time.

Store the returned NodeResult in the context for downstream nodes.

Persist execution details to tb_flow_log.

After all nodes succeed, return a FlowResult with overall success flag.

private PbcNode resolveNode(String nodeCode) { ... }

private void applyMappings(FlowContext context, FlowDefinition.NodeStep step) { ... }

Key Annotations

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
public @interface BusinessObject {
    String module();
    String nodeCode();
    String nodeName();
    String description() default "";
}

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface ExternalRestrictions {
    String[] requiredFields() default {};
}

PBC Node Interface

public interface PbcNode {
    NodeResult execute(FlowContext context);
}

Sample Node Implementations

Vehicle Arrival

@Component
@BusinessObject(module = "Car", nodeCode = "CAR_ARR_001", nodeName = "Vehicle Arrival")
public class CarArrivalBusinessObject implements PbcNode {
    @Autowired
    private ObjectMapper objectMapper;
    @Override
    @ExternalRestrictions
    public NodeResult execute(FlowContext context) {
        try {
            Object payload = context.getRequestPayload();
            String configJson = "{\"is_support_receipt\":true}";
            NodeConfiguration config = objectMapper.readValue(configJson, NodeConfiguration.class);
            Map<String, Object> result = new HashMap<>();
            result.put("status", "ARRIVED");
            result.put("arrivalTime", LocalDateTime.now().toString());
            if (Boolean.TRUE.equals(config.getIsSupportReceipt())) {
                result.put("receiptGenerated", true);
            }
            return NodeResult.ok("CAR_ARR_001", result);
        } catch (Exception e) {
            return NodeResult.fail("CAR_ARR_001", e.getMessage());
        }
    }
}

Transfer Sort

@Component
@BusinessObject(module = "Transfer", nodeCode = "TRANS_SORT_002", nodeName = "Transfer Sort")
public class TransferSortBusinessObject implements PbcNode {
    @Override
    public NodeResult execute(FlowContext context) {
        NodeResult upstream = context.getNodeOutput("CAR_ARR_001");
        Map<String, Object> arrivalData = (Map<String, Object>) upstream.getData();
        Map<String, Object> result = new HashMap<>();
        result.put("sortStatus", "SORTED");
        result.put("arrivalStatus", arrivalData.get("status"));
        return NodeResult.ok("TRANS_SORT_002", result);
    }
}

Vehicle Departure

@Component
@BusinessObject(module = "Car", nodeCode = "CAR_DEP_003", nodeName = "Vehicle Departure")
public class CarDepartureBusinessObject implements PbcNode {
    @Override
    public NodeResult execute(FlowContext context) {
        NodeResult upstream = context.getNodeOutput("TRANS_SORT_002");
        Map<String, Object> sortData = (Map<String, Object>) upstream.getData();
        Map<String, Object> result = new HashMap<>();
        result.put("departureStatus", "DEPARTED");
        result.put("sortStatus", sortData.get("sortStatus"));
        return NodeResult.ok("CAR_DEP_003", result);
    }
}

Database Schema

CREATE TABLE `tb_nodes` (
  `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
  `module` VARCHAR(255) DEFAULT '',
  `node_code` VARCHAR(100) NOT NULL,
  `node_name` VARCHAR(255) DEFAULT '',
  `node_description` VARCHAR(255) DEFAULT '',
  `configuration` JSON,
  `restrictions` JSON,
  `create_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  `update_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  `is_delete` SMALLINT NOT NULL DEFAULT 0,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_node_code` (`node_code`),
  KEY `idx_module` (`module`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

CREATE TABLE `tb_flow` (
  `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
  `process_name` VARCHAR(255) NOT NULL,
  `flow_description` VARCHAR(1024) NOT NULL,
  `nodes` JSON,
  `conditions` JSON,
  `create_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  `update_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  `is_delete` SMALLINT NOT NULL DEFAULT 0,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

CREATE TABLE `tb_flow_log` (
  `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
  `flow_id` BIGINT UNSIGNED,
  `flow_name` VARCHAR(255) DEFAULT '',
  `node_code` VARCHAR(100) DEFAULT '',
  `node_order` INT DEFAULT 0,
  `status` VARCHAR(20) DEFAULT '',
  `request_data` JSON,
  `response_data` JSON,
  `error_message` VARCHAR(2000) DEFAULT '',
  `duration_ms` INT DEFAULT 0,
  `create_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  KEY `idx_flow_id` (`flow_id`),
  KEY `idx_node_code` (`node_code`),
  KEY `idx_create_time` (`create_time`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

Metadata Scanning and Registration

@Component
public class PbcMetadataScanner implements ApplicationRunner {
    @Autowired private ApplicationContext applicationContext;
    @Autowired private BusinessObjectRegistry registry;
    @Override
    public void run(ApplicationArguments args) {
        Map<String, Object> beans = applicationContext.getBeansWithAnnotation(BusinessObject.class);
        for (Object bean : beans.values()) {
            BusinessObject ann = bean.getClass().getAnnotation(BusinessObject.class);
            BusinessObjectDTO dto = new BusinessObjectDTO();
            dto.setModule(ann.module());
            dto.setNodeCode(ann.nodeCode());
            dto.setNodeName(ann.nodeName());
            dto.setDescription(ann.description());
            registry.register(dto);
        }
    }
}

@Component
public class BusinessObjectRegistry {
    private final Map<String, BusinessObjectDTO> nodeCache = new ConcurrentHashMap<>();
    public void register(BusinessObjectDTO nodeDTO) { nodeCache.put(nodeDTO.getNodeCode(), nodeDTO); }
    public List<BusinessObjectDTO> getAllNodes() { return new ArrayList<>(nodeCache.values()); }
    public BusinessObjectDTO getNodeByCode(String nodeCode) { return nodeCache.get(nodeCode); }
}

@RestController
@RequestMapping("/api/pbc/metadata")
public class PbcMetadataController {
    @Autowired private BusinessObjectRegistry registry;
    @GetMapping("/nodes")
    public List<BusinessObjectDTO> getAllNodes() { return registry.getAllNodes(); }
}

AOP Aspect for External Restrictions

@Aspect
@Component
public class PbcExecutionAspect {
    @Autowired private ObjectMapper objectMapper;
    @Autowired private NodeMapper nodeMapper;
    @Around("@annotation(externalRestrictions)")
    public Object processBusinessObject(ProceedingJoinPoint joinPoint, ExternalRestrictions externalRestrictions) throws Throwable {
        BusinessObject bo = joinPoint.getTarget().getClass().getAnnotation(BusinessObject.class);
        String restrictionsJson = nodeMapper.findRestrictionsByNodeCode(bo.nodeCode());
        if (restrictionsJson == null || restrictionsJson.isEmpty()) {
            return joinPoint.proceed();
        }
        NodeRestrictions restrictions = objectMapper.readValue(restrictionsJson, NodeRestrictions.class);
        Object[] args = joinPoint.getArgs();
        if (args.length > 0 && args[0] != null) {
            JsonNode requestJson = objectMapper.valueToTree(args[0]);
            if (!evaluateRestrictions(restrictions, requestJson)) {
                throw new RuntimeException("外部约束条件校验失败：报文数据不满足规则");
            }
        }
        return joinPoint.proceed();
    }
    private boolean evaluateRestrictions(NodeRestrictions restrictions, JsonNode requestJson) {
        if (restrictions == null || restrictions.getRules() == null || restrictions.getRules().isEmpty()) return true;
        boolean isAnd = "AND".equalsIgnoreCase(restrictions.getLogicalOperator());
        boolean result = isAnd;
        for (NodeRestrictions.Rule rule : restrictions.getRules()) {
            JsonNode fieldNode = requestJson.path(rule.getField());
            String actual = fieldNode.isMissingNode() ? null : fieldNode.asText();
            boolean matched = "eq".equalsIgnoreCase(rule.getOperator()) && rule.getValue().equals(actual);
            if (isAnd) { result = result && matched; if (!result) break; }
            else { result = result || matched; if (result) break; }
        }
        return result;
    }
}

Mapper Interfaces

@Mapper
public interface FlowMapper {
    @Select("SELECT id, process_name, flow_description, nodes, conditions FROM tb_flow WHERE is_delete = 0")
    List<FlowDefinition> findAllActive();
}

@Mapper
public interface NodeMapper {
    @Select("SELECT restrictions FROM tb_nodes WHERE node_code = #{nodeCode} AND is_delete = 0")
    String findRestrictionsByNodeCode(String nodeCode);
}

Core Design Decisions

Execution entry : unified /api/pbc/flow/execute so all scenarios share one API; routing is performed by the flow engine.

Flow matching : condition‑based (state‑machine style) matching; each combination must be unique, validated on configuration.

Node data exchange : explicit pipeline‑style mapping; upstream output is read from FlowContext by downstream nodes.

Data loading / persistence : handled centrally by the engine; nodes focus only on business logic.

Exception handling : fast‑fail – any node failure aborts the whole flow.

Configuration scope : node‑internal switches only affect that node, not the overall flow path.

ExternalRestrictions scope : node‑level validation, separate from flow routing.

Condition Matching Details

Each flow’s conditions JSON contains a logical operator and a list of rules. Supported operators are: eq: equals (e.g., country eq CN) ne: not equals in: value in a list (e.g., region in [华东,华南]) gt: greater than (numeric) lt: less than (numeric)

{
  "logical_operator": "AND",
  "rules": [
    {"field": "country", "operator": "eq", "value": "CN"},
    {"field": "region", "operator": "eq", "value": "华东"},
    {"field": "warehouse", "operator": "eq", "value": "SH01"}
  ]
}

Project Structure (key modules)

pbc-application

: Spring Boot starter. pbc-common: annotations and utilities. pbc-engine: flow matcher, executor, controller. pbc-datasource: MyBatis‑Plus DAOs. pbc-nodes: concrete PBC node implementations (e.g., CarArrivalBusinessObject, TransferSortBusinessObject, CarDepartureBusinessObject).

pom.xml (excerpt)

<project xmlns="http://maven.apache.org/POM/4.0.0">
  <modelVersion>4.0.0</modelVersion>
  <parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.7.18</version>
  </parent>
  <groupId>com.pbc</groupId>
  <artifactId>pbc-demo</artifactId>
  <version>1.0.0</version>
  <properties>
    <java.version>11</java.version>
  </properties>
  <dependencies>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-aop</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-data-jpa</artifactId>
    </dependency>
    <dependency>
      <groupId>com.mysql</groupId>
      <artifactId>mysql-connector-j</artifactId>
      <scope>runtime</scope>
    </dependency>
    <dependency>
      <groupId>org.projectlombok</groupId>
      <artifactId>lombok</artifactId>
      <optional>true</optional>
    </dependency>
    <dependency>
      <groupId>com.fasterxml.jackson.core</groupId>
      <artifactId>jackson-databind</artifactId>
    </dependency>
  </dependencies>
  <build>
    <plugins>
      <plugin>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-maven-plugin</artifactId>
      </plugin>
    </plugins>
  </build>
</project>

End‑to‑End Example (East China SH01 Warehouse)

Configuration Data

Insert node metadata into tb_nodes:

INSERT INTO tb_nodes (module, node_code, node_name, node_description, configuration, restrictions) VALUES
('Car', 'CAR_ARR_001', '车辆到货环节', '处理车辆到货逻辑', '{"is_support_receipt": true}', '{"logical_operator":"AND","rules":[{"field":"areaType","operator":"eq","value":"V0055"}]}'),
('Transfer', 'TRANS_SORT_002', '中转分拨环节', '处理包裹分拨逻辑', '{"auto_sort": true}', NULL),
('Car', 'CAR_DEP_003', '车辆发货环节', '处理车辆发货逻辑', NULL, NULL);

Insert flow definition into tb_flow:

INSERT INTO tb_flow (process_name, flow_description, nodes, conditions) VALUES (
  '华东地区标准流程',
  '中国华东地区 SH01 仓库中心的标准处理流程',
  '[
    {"nodeCode":"CAR_ARR_001","order":1,"mappings":[]},
    {"nodeCode":"TRANS_SORT_002","order":2,"mappings":[{"fromField":"status","toField":"arrivalStatus"}]},
    {"nodeCode":"CAR_DEP_003","order":3,"mappings":[{"fromField":"sortStatus","toField":"inputSortStatus"}]}
  ]',
  '{"logical_operator":"AND","rules":[{"field":"country","operator":"eq","value":"CN"},{"field":"region","operator":"eq","value":"华东"},{"field":"warehouse","operator":"eq","value":"SH01"}]}'
);

Invocation

POST /api/pbc/flow/execute
Content-Type: application/json

{
  "country": "CN",
  "region": "华东",
  "warehouse": "SH01",
  "areaType": "V0055",
  "orderId": "ORD-20260430-001"
}

Execution flow:

FlowMatcher matches the request to the "华东地区标准流程" based on country/region/warehouse.

FlowExecutor runs three nodes in order. CAR_ARR_001: ExternalRestrictions validates areaType=V0055, then returns {"status":"ARRIVED",...}. TRANS_SORT_002: reads status from previous output, returns {"sortStatus":"SORTED","arrivalStatus":"ARRIVED"}. CAR_DEP_003: reads sortStatus, returns {"departureStatus":"DEPARTED","sortStatus":"SORTED"}.

Result JSON:

{
  "flowId": 1,
  "flowName": "华东地区标准流程",
  "success": true,
  "nodeResults": [
    {"nodeCode":"CAR_ARR_001","success":true,"data":{"status":"ARRIVED","arrivalTime":"2026-04-30T16:30:00","receiptGenerated":true}},
    {"nodeCode":"TRANS_SORT_002","success":true,"data":{"sortStatus":"SORTED","arrivalStatus":"ARRIVED"}},
    {"nodeCode":"CAR_DEP_003","success":true,"data":{"departureStatus":"DEPARTED","sortStatus":"SORTED"}}
  ]
}