算法艺术创作
algorithmic-art
通过代码和可调参数创作生成艺术与交互式视觉效果。
enterprise-integration-patterns【设计与创意】Use this skill when designing enterprise system integrations with message routing, protocol transformation, and error handling. This skill enforces: anti-corruption layers, SLA-defined integrations, dead letter queue monitoring. Do NOT use for: in-process function calls, simple REST API clients, database replication tools.
独立 GitHub 快照 · 自动分类 · 以原作者文档为准
Designs enterprise integration systems with message routing, protocol transformation, and error handling.
A six-phase methodology for enterprise integration:
Phase 1 - Identify: Map all systems requiring integration. Define system-of-record per data domain. Document protocols, data formats, and latency requirements. Identify integration touchpoints and data flows.
Phase 2 - Negotiate: Define contracts between systems (API specs, message schemas, SLA terms). Establish consumer-driven contracts. Agree on error handling and retry policies.
Phase 3 - Transform: Build anti-corruption layers between bounded contexts. Implement protocol and data format transformations. Enforce schema validation at boundaries.
Phase 4 - Route: Design message routing topology. Select integration style (API/messaging/streaming/file). Configure content-based, header-based, or topic-based routing.
Phase 5 - Govern: Monitor SLAs per integration flow. Track DLQ depth and error rates. Implement end-to-end tracing. Manage schema evolution with compatibility checks.
Phase 6 - Retire: Decommission deprecated integration points. Migrate consumers to new endpoints. Archive integration documentation.
Messaging Patterns:
Routing Patterns:
Transformation Patterns:
Is real-time response required?
├── Yes → Is strong consistency needed?
│ ├── Yes → API (REST/gRPC) synchronous
│ └── No → Can consumer tolerate eventual consistency?
│ ├── Yes → Messaging (async, durable)
│ └── No → Streaming (Kafka, Kinesis)
└── No → Is it a bulk data transfer?
├── Yes → File Transfer (SFTP, S3)
└── No → Messaging (async, queue-based)
| Style | Latency | Consistency | Volume | Best For |
|---|---|---|---|---|
| API (REST/gRPC) | Low | Strong | Low-Med | Real-time, CRUD, synchronous |
| Messaging (RabbitMQ, SQS) | Medium | Eventual | Medium | Async, durable, decoupled |
| Streaming (Kafka, Kinesis) | Low | Eventual | Very High | Real-time events, log processing |
| File Transfer (SFTP, S3) | High | Eventual | High | Batch, legacy systems |
| Database Sharing | Low | Strong | High | Shared schema (avoid if possible) |
Is throughput > 500K messages/second?
├── Yes → Apache Kafka (high-throughput, ordered per partition)
└── No → Are complex routing rules needed?
├── Yes → RabbitMQ (flexible exchanges, routing keys)
└── No → Is it fully on AWS?
├── Yes → SQS (simple, unlimited scale)
└── No → RabbitMQ (balanced features)
Do you need message replay/reprocessing?
├── Yes → Apache Kafka (retain and replay)
└── No → RabbitMQ or SQS (consume and remove)
| Topology | Flexibility | Complexity | Maintenance |
|---|---|---|---|
| Point-to-Point | Low | Low | High (N*N connections) |
| Hub-and-Spoke (ESB) | Medium | Medium | Medium (single broker) |
| Message Bus | High | High | Medium (routing rules) |
| Event Mesh | Very High | Very High | High (distributed brokers) |
| Strategy | Use When |
|---|---|
| Facade | Legacy system API is complex or inconsistent |
| Adapter | Legacy API differs from modern contract |
| Translator | Data models differ between systems |
| Event Translation | Events from legacy use different schema |
Exact user phrases: enterprise integration, system integration, legacy integration, ESB, message broker, data sync, system-of-record, integration pattern, API gateway integration, legacy system, middleware, event-driven integration, point-to-point, publish-subscribe.
Integration architecture design with routing, transformation, error handling, and monitoring.
## Integration Architecture
### Systems: {sources} -> {targets}
### Style: {API / messaging / file / streaming}
### Message Flow
{source} -> {transform} -> {route} -> {target}
### Routing Rules
{content / header / topic / rule-based}
### Error Handling
{retry: strategy, DLQ: location, alert: threshold}
### Monitoring
{tracing, SLAs, dashboards}
No preamble. No postamble. No explanations.
7000 tokens
Choose from API (synchronous, request-response), messaging (async, durable, pub-sub), file transfer (batch, CSV/JSON/EDI), database sharing (schema federation, views), or streaming (real-time, Kafka/Kinesis). Base decision on latency requirements, data volume, consistency needs, and system autonomy.
Place ACL between bounded contexts. Translate domain models at boundaries. Prevent internal model changes from propagating to external systems. Implement with separate translation layer, DTO objects, and interface adapters.
Apply content-based routing (message content determines destination), header-based (metadata-driven), topic-based (pub-sub per topic), or rule-based (complex condition evaluation). Route to single or multiple consumers.
Transform between SOAP/WSDL to REST/JSON, fixed-width files to structured payloads, EDI X12 to API objects, protobuf to Avro. Schema validation at boundaries. Version negotiation.
Implement retry with exponential backoff (base delay 1s, max 30s, jitter 0.1). Dead letter queue for poison messages. Manual intervention queue for business errors. Idempotency keys for safe retries.
End-to-end distributed tracing with trace ID propagation. SLA tracking (latency, throughput, error rate). Throughput dashboards partitioned by integration flow. Alert on DLQ depth, latency spikes, error rate thresholds.
Direct database access between services creates tight coupling. Schema changes in one service break others. Always use APIs or message queues for service-to-service communication. Database sharing is an anti-pattern in microservice architectures. If unavoidable, use views or read replicas.
Integrating directly with a legacy system's internal model propagates legacy problems. Changes to the legacy system break the integration. ACL isolates internal domain from external models. Always implement ACL between bounded contexts.
Service A calls B calls C calls D. Latency = sum of all. Failure of any breaks the entire chain. Use async messaging for non-real-time processing. Implement circuit breakers for synchronous calls. Consider CQRS to decouple read and write paths.
Messages that fail processing go to DLQ but nobody monitors it. DLQ fills up, retention overflows, messages lost. Monitor DLQ depth. Alert on growth. Have reprocessing mechanism. Review DLQ content weekly.
Retry without idempotency causes duplicate orders, payments, or data entries. All state-changing operations must be idempotent. Use idempotency keys. De-duplicate on consumer side. Store processed message IDs.
SOAP service expects XML; new microservice sends JSON. Transformation layer is missing. Always plan for protocol transformation at integration boundaries. Use message brokers with built-in transformation (Kafka Connect, Debezium) or custom transformer.
When a downstream service fails, the caller keeps retrying and exacerbates the issue. Circuit breaker stops calls when failure rate exceeds threshold, giving the downstream time to recover. Implement circuit breakers for all synchronous integrations.
Producer changes message schema; consumer fails to deserialize. Always enforce schema compatibility checks. Use schema registry (Confluent Schema Registry, Apicurio). Set compatibility policy: backward (default), forward, or full.
API Gateway: synchronous request handling, routing, rate limiting, auth. Message Broker: async pub/sub, queue, durable messaging. ESB: heavy middleware with routing, transformation, orchestration, governance. Modern approach: API Gateway for sync, Message Broker for async, avoid heavy ESB. Decompose ESB capabilities into lighter components.
Kafka: high-throughput streaming, event log, replay, strong ordering per partition, consumer groups. RabbitMQ: flexible routing, reliable task queues, RPC, lower throughput but simpler ops. Kafka for event-driven and streaming. RabbitMQ for task queues and RPC. Many orgs use both for different use cases.
Integration: connecting systems, message routing, protocol transformation. Orchestration: coordinating a business process across multiple systems (e.g., order fulfillment: payment -> inventory -> shipping). Orchestration uses integration as infrastructure. Use workflow engines (Camunda, Temporal) for orchestration; use integration patterns for system connectivity.
from abc import ABC, abstractmethod
from dataclasses import dataclass, asdict
import json
# Legacy system domain model
@dataclass
class LegacyOrder:
order_id: str
customer_code: str
item_list: str # Comma-separated
total_dollars: float
created_dt: str # MM/DD/YYYY string
# Modern system domain model
@dataclass
class ModernOrder:
id: str
customer_id: str
items: list[str]
total_cents: int
created_at: str # ISO 8601
class AntiCorruptionLayer(ABC):
@abstractmethod
def translate(self, legacy: LegacyOrder) -> ModernOrder:
pass
class OrderTranslator(AntiCorruptionLayer):
def translate(self, legacy: LegacyOrder) -> ModernOrder:
from datetime import datetime
return ModernOrder(
id=legacy.order_id,
customer_id=legacy.customer_code,
items=[i.strip() for i in legacy.item_list.split(",")],
total_cents=int(legacy.total_dollars * 100),
created_at=datetime.strptime(legacy.created_dt, "%m/%d/%Y").isoformat()
)
# Facade for legacy API
class LegacyOrderFacade:
def __init__(self, legacy_api_url):
self.api_url = legacy_api_url
def get_order(self, order_id: str) -> LegacyOrder:
# Simulated legacy API call
return LegacyOrder(
order_id=order_id,
customer_code="CUST001",
item_list="Widget A, Widget B, Gadget C",
total_dollars=149.99,
created_dt="01/15/2025"
)
# Modern consumer
class OrderService:
def __init__(self, translator: AntiCorruptionLayer, legacy_facade: LegacyOrderFacade):
self.translator = translator
self.legacy_facade = legacy_facade
def get_modern_order(self, order_id: str) -> dict:
legacy = self.legacy_facade.get_order(order_id)
modern = self.translator.translate(legacy)
return asdict(modern)
import json, enum
from typing import Callable
class Message:
def __init__(self, body: dict, headers: dict = None):
self.body = body
self.headers = headers or {}
class RoutingRule:
def __init__(self, name: str, condition: Callable[[Message], bool], destination: str):
self.name = name
self.condition = condition
self.destination = destination
class ContentBasedRouter:
def __init__(self):
self.rules: list[RoutingRule] = []
def add_rule(self, rule: RoutingRule):
self.rules.append(rule)
def route(self, message: Message) -> list[str]:
destinations = []
for rule in self.rules:
if rule.condition(message):
destinations.append(rule.destination)
return destinations or ["dead-letter-queue"]
# Example usage
router = ContentBasedRouter()
router.add_rule(RoutingRule(
name="high-value-orders",
condition=lambda m: m.body.get("amount", 0) > 10000,
destination="order-fulfillment-premium"
))
router.add_rule(RoutingRule(
name="international-shipments",
condition=lambda m: m.body.get("region") == "international",
destination="shipping-international"
))
router.add_rule(RoutingRule(
name="fraud-review",
condition=lambda m: m.body.get("risk_score", 0) > 0.8,
destination="fraud-detection"
))
msg = Message({"amount": 15000, "region": "domestic", "risk_score": 0.3})
print(router.route(msg)) # ['order-fulfillment-premium']
import time, random
from functools import wraps
def retry_with_backoff(max_retries=3, base_delay=1.0, max_delay=30.0, jitter=True):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_retries + 1):
try:
return func(*args, **kwargs)
except Exception as e:
last_exception = e
if attempt < max_retries:
delay = min(base_delay * (2 ** attempt), max_delay)
if jitter:
delay = delay * (0.5 + random.random() * 0.5)
time.sleep(delay)
raise last_exception
return wrapper
return decorator
@retry_with_backoff(max_retries=3, base_delay=1.0)
def call_downstream_service(url):
# Simulated API call
response = requests.get(url, timeout=5)
response.raise_for_status()
return response.json()
class SchemaRegistry:
def __init__(self):
self.schemas = {}
def register_schema(self, subject: str, schema: dict, compatibility: str = "backward"):
if subject in self.schemas:
existing = self.schemas[subject][-1]
if compatibility == "backward":
self._check_backward_compatibility(existing, schema)
elif compatibility == "forward":
self._check_forward_compatibility(existing, schema)
elif compatibility == "full":
self._check_backward_compatibility(existing, schema)
self._check_forward_compatibility(existing, schema)
self.schemas.setdefault(subject, []).append(schema)
def _check_backward_compatibility(self, existing, new_schema):
existing_fields = {f["name"]: f for f in existing.get("fields", [])}
for field in new_schema.get("fields", []):
if field["name"] in existing_fields:
if field.get("type") != existing_fields[field["name"]].get("type"):
raise ValueError(f"Field {field['name']} type changed from "
f"{existing_fields[field['name']]['type']} to {field['type']}")
print("Backward compatibility check passed")
def _check_forward_compatibility(self, existing, new_schema):
new_field_names = {f["name"] for f in new_schema.get("fields", [])}
for field in existing.get("fields", []):
if field["name"] not in new_field_names:
if not field.get("optional", False):
raise ValueError(f"Field {field['name']} removed but not optional")
print("Forward compatibility check passed")
registry = SchemaRegistry()
registry.register_schema("order-value", {
"type": "record",
"fields": [
{"name": "order_id", "type": "string"},
{"name": "amount", "type": "float"}
]
})
registry.register_schema("order-value", {
"type": "record",
"fields": [
{"name": "order_id", "type": "string"},
{"name": "amount", "type": "float"},
{"name": "currency", "type": "string", "default": "USD"}
]
})
Every system connects directly to every other system. N systems create N² connections. Changes ripple across the entire network. Use a message broker or API gateway as a central routing point. Limit direct connections to stable, high-throughput paths.
Routing, transformation, orchestration, protocol conversion, business rules — all in one ESB. The ESB becomes a monolith that is harder to change than any of the systems it connects. Decompose into discrete: message broker, API gateway, transformation service, workflow engine.
Service A calls B calls C calls D. Latency equals the sum of all four. Failure of D takes down A, B, and C. Use async messaging between non-real-time steps. Implement circuit breakers and timeouts. Consider CQRS to separate read/write.
Producer changes a message schema without notifying consumers. Consumers break silently. Always use a schema registry with compatibility checks. Notify consumer teams before schema changes. Version all schemas.
Messages fail, go to DLQ, but nobody monitors it. DLQ messages accumulate and eventually overflow retention. Critical business events are permanently lost. Monitor DLQ depth with alerts. Review DLQ content weekly. Have a reprocessing workflow.
For monitoring integration SLAs, hand off to enterprise-sla-management. For data governance across integrations, hand off to enterprise-data-governance.