Event-Mesh-Architektur für Multi-Cloud-Systeme
Entwirf Event-Mesh-Architekturen, die Events über Clouds und Regionen routen: dynamisches Topic-Routing, Protokoll-Bridging, Schema-Governance.
Ein Event-Broker funktioniert gut, solange alles in einem Cluster läuft. Aber moderne Systeme erstrecken sich über mehrere Clouds, Regionen und lokale Datacenter. Ein Event Mesh erweitert das event-driven Pattern über diese Grenzen hinaus und schafft einen Stoff, in dem jeder Producer jeden Consumer erreichen kann, egal wo er läuft. Im Gegensatz zu einem zentralisierten Broker ist das Mesh verteilt – jeder Node verarbeitet lokalen Traffic und routed grenzüberschreitende Events dynamisch.
Die Herausforderungen gehen über Netzwerkfragen hinaus. Du brauchst konsistente Schemas über Umgebungen hinweg, Protokollübersetzung zwischen verschiedenen Messaging-Systemen und Delivery-Garantien, die regionale Ausfälle überstehen.
Mesh-Topologie und dynamisches Routing
Ein Event Mesh verbindet Broker über Regionen hinweg mit intelligentem Routing. Events fließen zu Consumern basierend auf Topic-Subscriptions, nicht auf hartcodierten Zielen.
// ❌ Point-to-point integration — doesn't scale
const routes = {
"us-east": "kafka://us-east-broker:9092",
"eu-west": "kafka://eu-west-broker:9092",
"ap-south": "rabbitmq://ap-south-broker:5672",
};
// Adding a new region means updating every producer
// Different protocols? Manual translation at each hop// ✅ Event mesh with dynamic routing
interface MeshNode {
id: string;
region: string;
cloud: string;
protocol: "kafka" | "amqp" | "mqtt" | "nats";
topics: Set<string>;
connectedPeers: Set<string>;
}
interface RouteEntry {
topic: string;
targetNodes: string[];
priority: number;
ttl: number;
}
class EventMeshRouter {
private nodes: Map<string, MeshNode> = new Map();
private routingTable: Map<string, RouteEntry[]> = new Map();
registerNode(node: MeshNode): void {
this.nodes.set(node.id, node);
this.rebuildRoutes();
}
subscribe(nodeId: string, topic: string): void {
const node = this.nodes.get(nodeId);
if (!node) return;
node.topics.add(topic);
this.updateRoutesForTopic(topic);
}
route(
topic: string,
sourceNode: string
): MeshNode[] {
const routes = this.routingTable.get(topic) ?? [];
return routes
.flatMap((route) =>
route.targetNodes
.filter((id) => id !== sourceNode)
.map((id) => this.nodes.get(id))
.filter((n): n is MeshNode => n !== undefined)
);
}
private rebuildRoutes(): void {
this.routingTable.clear();
for (const [nodeId, node] of this.nodes) {
for (const topic of node.topics) {
this.updateRoutesForTopic(topic);
}
}
}
private updateRoutesForTopic(topic: string): void {
const subscribedNodes: string[] = [];
for (const [nodeId, node] of this.nodes) {
if (node.topics.has(topic)) {
subscribedNodes.push(nodeId);
}
}
this.routingTable.set(topic, [
{
topic,
targetNodes: subscribedNodes,
priority: 1,
ttl: 300,
},
]);
}
}Wenn eine neue Region online geht, registriert sich ihr Broker im Mesh, abonniert Topics und empfängt sofort Events. Keine Producer-Änderungen nötig.
Protokoll-Bridging
Verschiedene Umgebungen nutzen verschiedene Messaging-Protokolle. Das Mesh muss transparent zwischen ihnen übersetzen.
interface MeshEvent {
id: string;
topic: string;
payload: Uint8Array;
contentType: string;
headers: Map<string, string>;
timestamp: number;
sourceNode: string;
schemaVersion: string;
}
interface ProtocolBridge {
protocol: string;
serialize(event: MeshEvent): Uint8Array;
deserialize(data: Uint8Array, metadata: Record<string, string>): MeshEvent;
}
class KafkaBridge implements ProtocolBridge {
protocol = "kafka";
serialize(event: MeshEvent): Uint8Array {
const record = {
key: event.id,
value: event.payload,
headers: Object.fromEntries(event.headers),
timestamp: event.timestamp,
};
return new TextEncoder().encode(JSON.stringify(record));
}
deserialize(
data: Uint8Array,
metadata: Record<string, string>
): MeshEvent {
const record = JSON.parse(new TextDecoder().decode(data));
return {
id: record.key,
topic: metadata.topic ?? "",
payload: new TextEncoder().encode(
typeof record.value === "string"
? record.value
: JSON.stringify(record.value)
),
contentType: "application/json",
headers: new Map(Object.entries(record.headers ?? {})),
timestamp: record.timestamp ?? Date.now(),
sourceNode: metadata.sourceNode ?? "",
schemaVersion: record.headers?.["schema-version"] ?? "1.0",
};
}
}
class AmqpBridge implements ProtocolBridge {
protocol = "amqp";
serialize(event: MeshEvent): Uint8Array {
const message = {
messageId: event.id,
body: event.payload,
applicationProperties: Object.fromEntries(event.headers),
creationTime: event.timestamp,
};
return new TextEncoder().encode(JSON.stringify(message));
}
deserialize(
data: Uint8Array,
metadata: Record<string, string>
): MeshEvent {
const message = JSON.parse(new TextDecoder().decode(data));
return {
id: message.messageId,
topic: metadata.topic ?? "",
payload: message.body,
contentType: "application/json",
headers: new Map(
Object.entries(message.applicationProperties ?? {})
),
timestamp: message.creationTime ?? Date.now(),
sourceNode: metadata.sourceNode ?? "",
schemaVersion:
message.applicationProperties?.["schema-version"] ?? "1.0",
};
}
}
class BridgeManager {
private bridges: Map<string, ProtocolBridge> = new Map();
register(bridge: ProtocolBridge): void {
this.bridges.set(bridge.protocol, bridge);
}
translate(
event: MeshEvent,
sourceProtocol: string,
targetProtocol: string
): Uint8Array {
const targetBridge = this.bridges.get(targetProtocol);
if (!targetBridge) {
throw new Error(
`No bridge for protocol: ${targetProtocol}`
);
}
return targetBridge.serialize(event);
}
}Schema-Governance im Mesh
Events, die zwischen unabhängig deployten Services fließen, brauchen Schema-Verträge. Ohne Governance driften Producer und Consumer stillschweigend auseinander.
interface SchemaDefinition {
id: string;
topic: string;
version: number;
schema: object;
compatibility: "backward" | "forward" | "full" | "none";
registeredAt: Date;
}
class MeshSchemaRegistry {
private schemas: Map<string, SchemaDefinition[]> = new Map();
register(schema: SchemaDefinition): {
accepted: boolean;
reason?: string;
} {
const existing = this.schemas.get(schema.topic) ?? [];
const latest = existing[existing.length - 1];
if (latest) {
const compatible = this.checkCompatibility(
latest,
schema
);
if (!compatible.ok) {
return {
accepted: false,
reason: compatible.reason,
};
}
}
existing.push(schema);
this.schemas.set(schema.topic, existing);
return { accepted: true };
}
getLatest(topic: string): SchemaDefinition | null {
const versions = this.schemas.get(topic);
return versions?.[versions.length - 1] ?? null;
}
private checkCompatibility(
existing: SchemaDefinition,
proposed: SchemaDefinition
): { ok: boolean; reason?: string } {
switch (existing.compatibility) {
case "backward":
// New schema can read old data
return this.isBackwardCompatible(existing, proposed);
case "forward":
// Old schema can read new data
return this.isForwardCompatible(existing, proposed);
case "full":
// Both directions
const backward = this.isBackwardCompatible(
existing, proposed
);
const forward = this.isForwardCompatible(
existing, proposed
);
return {
ok: backward.ok && forward.ok,
reason: backward.reason ?? forward.reason,
};
case "none":
return { ok: true };
}
}
private isBackwardCompatible(
existing: SchemaDefinition,
proposed: SchemaDefinition
): { ok: boolean; reason?: string } {
// Simplified: check no required fields were removed
const existingFields = this.extractRequiredFields(
existing.schema
);
const proposedFields = this.extractRequiredFields(
proposed.schema
);
for (const field of existingFields) {
if (!proposedFields.has(field)) {
return {
ok: false,
reason: `Required field "${field}" removed`,
};
}
}
return { ok: true };
}
private isForwardCompatible(
existing: SchemaDefinition,
proposed: SchemaDefinition
): { ok: boolean; reason?: string } {
// New fields must have defaults
const existingFields = this.extractAllFields(existing.schema);
const proposedFields = this.extractAllFields(proposed.schema);
const newFields = [...proposedFields].filter(
(f) => !existingFields.has(f)
);
const requiredNew = this.extractRequiredFields(
proposed.schema
);
for (const field of newFields) {
if (requiredNew.has(field)) {
return {
ok: false,
reason: `New required field "${field}" without default`,
};
}
}
return { ok: true };
}
private extractRequiredFields(schema: object): Set<string> {
const s = schema as any;
return new Set(s.required ?? []);
}
private extractAllFields(schema: object): Set<string> {
const s = schema as any;
return new Set(Object.keys(s.properties ?? {}));
}
}Regionenübergreifende Delivery-Garantien
Events, die regionale Grenzen überschreiten, sehen Netzwerkpartitionen, Latenzspitzen und Ordering-Herausforderungen. Das Mesh muss vorhersehbare Delivery-Semantik bieten.
interface DeliveryConfig {
guaranty: "at-most-once" | "at-least-once" | "exactly-once";
maxRetries: number;
retryBackoff: number;
deduplicationWindow: number;
orderingKey?: string;
}
class CrossRegionDelivery {
private delivered: Map<string, number> = new Map();
async deliver(
event: MeshEvent,
targets: MeshNode[],
config: DeliveryConfig
): Promise<DeliveryReport> {
const results: DeliveryResult[] = [];
for (const target of targets) {
const result = await this.deliverToNode(
event,
target,
config
);
results.push(result);
}
return {
eventId: event.id,
totalTargets: targets.length,
delivered: results.filter((r) => r.success).length,
failed: results.filter((r) => !r.success),
};
}
private async deliverToNode(
event: MeshEvent,
target: MeshNode,
config: DeliveryConfig
): Promise<DeliveryResult> {
// Deduplication check
if (config.guaranty !== "at-most-once") {
const deliveryKey = `${event.id}:${target.id}`;
const lastDelivery = this.delivered.get(deliveryKey);
if (
lastDelivery &&
Date.now() - lastDelivery < config.deduplicationWindow
) {
return {
targetNode: target.id,
success: true,
deduplicated: true,
};
}
}
// Retry with backoff
for (let attempt = 0; attempt <= config.maxRetries; attempt++) {
try {
await this.send(event, target);
const deliveryKey = `${event.id}:${target.id}`;
this.delivered.set(deliveryKey, Date.now());
return {
targetNode: target.id,
success: true,
attempts: attempt + 1,
};
} catch (error) {
if (attempt < config.maxRetries) {
const delay =
config.retryBackoff * Math.pow(2, attempt);
await new Promise((r) => setTimeout(r, delay));
}
}
}
return {
targetNode: target.id,
success: false,
attempts: config.maxRetries + 1,
error: "Max retries exceeded",
};
}
private async send(
event: MeshEvent,
target: MeshNode
): Promise<void> {
// Protocol-specific send logic
console.log(
`Delivering ${event.id} to ${target.id} via ${target.protocol}`
);
}
}
interface DeliveryResult {
targetNode: string;
success: boolean;
attempts?: number;
deduplicated?: boolean;
error?: string;
}
interface DeliveryReport {
eventId: string;
totalTargets: number;
delivered: number;
failed: DeliveryResult[];
}Wichtige Erkenntnisse
Ein Event Mesh erweitert event-driven Architecture über Cloud-, Regions- und Protokollgrenzen hinweg, indem verteilte Broker mit dynamischem Routing verbunden werden. Dynamisches Topic-basiertes Routing bedeutet, dass neue Regionen oder Services dem Mesh beitreten, indem sie sich registrieren und abonnieren – ohne Producer-Änderungen. Protokoll-Bridges übersetzen Events transparent zwischen Kafka, AMQP, MQTT und anderen Protokollen, sodass heterogene Umgebungen am selben Event-Stoff teilnehmen können. Schema-Governance im Mesh verhindert stillschweigenden Vertragsdrift: Registriere Schemas mit Kompatibilitätsprüfungen, die breaking changes ablehnen, bevor sie die Consumer erreichen. Regionenübergreifende Delivery erfordert explizite Garantien: Wähle zwischen at-most-once, at-least-once oder exactly-once basierend auf den Geschäftsanforderungen und implementiere Deduplizierungsfenster, um die unvermeidlichen Duplikate aus der Retry-Logik zu behandeln. Das Event Mesh ist kein einzelnes Produkt – es ist ein architektonisches Pattern, das Router, Bridges, Registries und Delivery-Garantien zu einem einheitlichen Stoff für verteilten Event-Flow komponiert.
