composable-commerce

组合式商务

概述

组合式商务是一种基于 MACH 原则（微服务、API 优先、云原生、无头）的架构方法。每项商务能力（购物车、商品目录、搜索、CMS、结账、忠诚度）都由各领域最佳的服务提供，而非依赖单体平台。服务通过 API 和事件通信，让团队可以替换或升级单项能力，而不影响系统其他部分。本技能涵盖 MACH 架构模式、服务集成、事件驱动协调，以及在生产环境中运行组合式技术栈的运维考量。

何时使用此技能

• 当单体平台（Magento、Salesforce CC）已无法适应团队规模或流量模式
• 当不同领域（商品目录、结账、忠诚度）的发布周期和所有权各不相同时
• 当你需要组合各领域最佳的供应商时，例如用 Algolia 负责搜索、Contentful 负责 CMS、commercetools 负责商务
• 当进入新市场，需要为不同地区使用不同的履约、定价或税务服务商时
• 当构建的平台必须通过单一后端支持多个店面（网站、移动端、实体店、B2B 门户）时

前提条件与平台说明

本技能适用于自定义／无头店面（Node.js、Python 或类似后端）。代码示例使用 TypeScript／Node.js，可适配任何技术栈。

Shopify：Shopify Hydrogen 是 Shopify 的无头框架。使用 Shopify 作为商务后端并搭配自定义前端，或将 Shopify 与其他各领域最佳的服务混用时，均可应用 MACH／组合式模式。 WooCommerce：WooCommerce 可通过其 REST API 和 WPGraphQL 充当无头后端。这些模式适用于将前端与 WordPress 解耦的场景。 Magento：Magento 的 GraphQL API 和 PWA Studio 支持无头架构。在 MACH 技术栈中将 Magento 用作后端服务时，可应用这些组合式模式。

你需要：

• Node.js 18+（或根据你的后端语言进行适配）
• PostgreSQL（或你偏好的关系型数据库）
• 搜索服务（Algolia、Elasticsearch 或 Typesense）
• Stripe 账户和 API 密钥
• 电子邮件发送服务（SendGrid、AWS SES 或 Postmark）

核心指引

1. 围绕商务能力设计服务边界

   典型的可组合商务技术栈会将各项能力拆分为独立服务：

   能力	示例服务
   商品目录	commercetools、Akeneo PIM、Salsify
   搜索与发现	Algolia、Elasticsearch、Constructor.io
   CMS / 内容	Contentful、Sanity、Storyblok
   购物车与结账	commercetools、Elastic Path、Medusa
   支付	Stripe、Adyen、Braintree
   税务	Avalara、TaxJar
   配送与履约	EasyPost、ShipBob、自定义 OMS
   客户身份	Auth0、Okta、Cognito
   忠诚度与促销	Talon.One、Voucherify
   电子邮件 / 通知	SendGrid、Customer.io

   定义限界上下文：每个服务拥有自己的数据，并且只能通过 API 对外提供数据。绝不要在服务之间共享数据库。

2. 实现 API 组合层（BFF）

   后端面向前端（BFF）将多个上游 API 聚合到单个请求中，并根据前端需求进行定制：

   // bff/src/routes/product-page.ts
   import {getProduct} from '../services/catalog';
   import {getSearchReviews} from '../services/reviews';
   import {getRecommendations} from '../services/recommendations';
   import {getInventory} from '../services/inventory';
   
   export async function productPageData(productId: string, customerId?: string) {
     // Parallel fetch from independent services
     const [product, inventory, recommendations] = await Promise.all([
       getProduct(productId),
       getInventory(productId),
       getRecommendations(productId, customerId),
     ]);
   
     // Sequential: reviews needs product.sku
     const reviews = await getSearchReviews(product.sku);
   
     return {
       product,
       inventory,
       recommendations,
       reviews,
     };
   }
   

   使用架构拼接或联邦将 BFF 暴露为 GraphQL API：

   import {buildHTTPExecutor} from '@graphql-tools/executor-http';
   import {stitchSchemas} from '@graphql-tools/stitch';
   
   const gatewaySchema = await stitchSchemas({
     subschemas: [
       { schema: await introspectSchema(catalogExecutor), executor: catalogExecutor },
       { schema: await introspectSchema(inventoryExecutor), executor: inventoryExecutor },
       { schema: await introspectSchema(reviewsExecutor), executor: reviewsExecutor },
     ],
   });
   

3. 使用事件驱动架构协调跨服务流程

   对于跨越多项能力的工作流（下单 → 预留库存 → 触发履约 → 发送电子邮件），服务应异步通信：

   // Order service publishes an event after checkout
   import {EventBridge} from '@aws-sdk/client-eventbridge';
   
   const eventBridge = new EventBridge({region: 'us-east-1'});
   
   async function publishOrderPlaced(order: Order) {
     await eventBridge.putEvents({
       Entries: [
         {
           Source: 'commerce.orders',
           DetailType: 'OrderPlaced',
           Detail: JSON.stringify({
             orderId: order.id,
             customerId: order.customerId,
             lineItems: order.lineItems,
             totalAmount: order.totalAmount,
             currency: order.currency,
           }),
           EventBusName: 'commerce-events',
         },
       ],
     });
   }
   
   // Inventory service subscribes and reserves stock
   // EventBridge rule routes OrderPlaced → Lambda → inventory-service
   export async function handler(event: EventBridgeEvent<'OrderPlaced', OrderPayload>) {
     const {orderId, lineItems} = event.detail;
     await reserveInventory(lineItems);
     await publishInventoryReserved(orderId);
   }
   

4. 为分布式事务实现 Saga 模式

   当多步骤工作流必须跨服务保持原子性时，使用编排式 Saga，并配合补偿事务：

   // Choreography-based saga for order fulfillment
   // Each service publishes events and reacts to others
   
   // Order Service
   on('CheckoutCompleted', async ({orderId, paymentIntentId}) => {
     await orders.create({orderId, status: 'pending_payment'});
     await publish('OrderCreated', {orderId, paymentIntentId});
   });
   
   // Payment Service
   on('OrderCreated', async ({orderId, paymentIntentId}) => {
     try {
       await capturePayment(paymentIntentId);
       await publish('PaymentCaptured', {orderId});
     } catch (err) {
       await publish('PaymentFailed', {orderId, reason: err.message});
     }
   });
   
   // Fulfillment Service
   on('PaymentCaptured', async ({orderId}) => {
     await fulfillment.schedule(orderId);
   });
   
   // Order Service — compensate on failure
   on('PaymentFailed', async ({orderId}) => {
     await orders.update(orderId, {status: 'payment_failed'});
     await releaseInventory(orderId);
     await notifyCustomer(orderId, 'payment_failed');
   });
   

5. 管理 API 版本控制和向后兼容性

   在可组合技术栈中，各服务独立演进。采用增量式版本控制策略：

   // Use content negotiation or URL versioning
   // GET /api/v2/products/:id
   // Accept: application/vnd.commerce.product.v2+json
   
   // Apply the Tolerant Reader pattern — ignore unknown fields
   interface ProductV1 {
     id: string;
     name: string;
     price: number;
   }
   
   // V2 adds fields — existing consumers still work
   interface ProductV2 extends ProductV1 {
     categories?: string[];
     attributes?: Record<string, string>;
     brand?: string;
   }
   
   // Use feature flags to roll out breaking changes
   async function getProduct(id: string, apiVersion: '1' | '2' = '1') {
     const product = await catalog.findById(id);
     return apiVersion === '2' ? toProductV2(product) : toProductV1(product);
   }
   

6. 实现断路器以增强韧性

   当某个服务性能下降时，防止整个技术栈发生级联故障：

   import CircuitBreaker from 'opossum';
   
   const inventoryCircuit = new CircuitBreaker(checkInventory, {
     timeout: 3000,          // Request timeout in ms
     errorThresholdPercentage: 50, // Open circuit if 50% of requests fail
     resetTimeout: 30000,    // Try again after 30 seconds
   });
   
   inventoryCircuit.fallback(() => ({
     available: true,        // Optimistic fallback — show as available
     quantity: null,         // Don't show exact quantity
   }));
   
   inventoryCircuit.on('open', () => {
     logger.warn('Inventory service circuit OPEN — using fallback');
     metrics.increment('circuit_breaker.inventory.open');
   });
   
   // Usage
   const stock = await inventoryCircuit.fire(productId);
   

示例

将 commercetools SDK 集成到商品目录中

import {createApiBuilderFromCtpClient} from '@commercetools/platform-sdk';
import {ClientBuilder} from '@commercetools/sdk-client-v2';

const ctpClient = new ClientBuilder()
  .withProjectKey(process.env.CTP_PROJECT_KEY!)
  .withClientCredentialsFlow({
    host: 'https://auth.us-central1.gcp.commercetools.com',
    projectKey: process.env.CTP_PROJECT_KEY!,
    credentials: {
      clientId: process.env.CTP_CLIENT_ID!,
      clientSecret: process.env.CTP_CLIENT_SECRET!,
    },
    scopes: ['manage_project:' + process.env.CTP_PROJECT_KEY],
    fetch,
  })
  .withHttpMiddleware({host: 'https://api.us-central1.gcp.commercetools.com', fetch})
  .build();

const apiRoot = createApiBuilderFromCtpClient(ctpClient).withProjectKey({
  projectKey: process.env.CTP_PROJECT_KEY!,
});

// Fetch product by slug
const product = await apiRoot
  .products()
  .get({queryArgs: {where: `slug(en="${slug}")`, expand: ['productType']}})
  .execute();

使用 Contentful CMS 丰富 Algolia 搜索结果

import algoliasearch from 'algoliasearch';
import {createClient as createContentfulClient} from 'contentful';

const algolia = algoliasearch(process.env.ALGOLIA_APP_ID!, process.env.ALGOLIA_SEARCH_KEY!);
const contentful = createContentfulClient({space: process.env.CONTENTFUL_SPACE_ID!, accessToken: process.env.CONTENTFUL_TOKEN!});

async function searchProducts(query: string, filters?: string) {
  // Search in Algolia for product IDs and structured data
  const {hits} = await algolia.initIndex('products').search<AlgoliaProduct>(query, {
    filters,
    attributesToRetrieve: ['objectID', 'name', 'price', 'contentfulEntryId'],
  });

  // Enrich with rich content from Contentful
  const contentEntryIds = hits.map(h => h.contentfulEntryId).filter(Boolean);
  const contentEntries = await contentful.getEntries({
    content_type: 'productPage',
    'sys.id[in]': contentEntryIds.join(','),
  });

  const contentMap = new Map(contentEntries.items.map(e => [e.sys.id, e]));

  return hits.map(hit => ({
    ...hit,
    content: contentMap.get(hit.contentfulEntryId),
  }));
}

最佳实践

• 定义清晰的数据所有权模型 — 每项数据都应有且仅有一个权威数据源；其他服务通过 API 读取数据，绝不直接写入
• 为最终一致性而设计 — 事件发生后，跨服务数据会暂时不一致；构建能够容忍这种情况的 UI 和工作流（例如，乐观库存、补偿事务）
• 使用 API 网关处理横切关注点 — 身份验证、速率限制、请求日志记录和 SSL 终止应由网关负责，而不是由每个服务各自处理
• 实现分布式追踪 — 使用 OpenTelemetry，并在所有服务调用中传递追踪 ID；这对于调试多服务请求故障至关重要
• 为事件设置版本 — 事件架构会随时间变化；在每个事件负载中加入 schemaVersion 字段，并维护向后兼容的消费者
• 使用服务网格处理服务间流量 — Istio 或 AWS App Mesh 等工具可在基础设施层处理双向 TLS、负载均衡和重试
• 从模块化单体开始 — 只有在团队所有权边界清晰且具备足够的运维成熟度后，才拆分为微服务；过早拆分会造成意外的复杂性

常见陷阱

问题	解决方案
分布式事务失败导致数据不一致	使用补偿事务实现 Saga；采用 outbox 模式，确保数据库写入后发布事件
服务延迟在关键路径上叠加	通过异步事件将非关键服务移出关键路径；为所有外部调用设置严格的超时和熔断器
API 契约变更导致下游使用方出错	使用消费者驱动的契约测试（Pact），以便在部署前发现破坏性变更
共享数据库产生隐性耦合	执行“一项服务，一个 Schema”原则；使用事件溯源或变更数据捕获，在服务间传播数据
调试跨越 10 项服务的故障	从一开始就使用 OpenTelemetry 实现分布式追踪；为每个用户请求分配唯一的追踪 ID，以关联日志

相关技能

• @commerce-api-gateway
• @saleor-development
• @shopify-hydrogen
• @webhook-architecture
• @flash-sale-scaling
• @edge-commerce