编排与编排：设计微服务中的分布式工作流程

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"

	"github.com/segmentio/kafka-go"
)

type PaymentProcessedEvent struct {
	OrderID string  `json:"order_id"`
	Amount  float64 `json:"amount"`
	Status  string  `json:"status"`
}

type InventoryReservedEvent struct {
	OrderID string `json:"order_id"`
	Status  string `json:"status"`
}

func main() {
	reader := kafka.NewReader(kafka.ReaderConfig{
		Brokers: []string{"localhost:9092"},
		Topic:   "payment-events",
		GroupID: "inventory-service-group",
	})
	defer reader.Close()

	writer := kafka.NewWriter(kafka.WriterConfig{
		Brokers: []string{"localhost:9092"},
		Topic:   "inventory-events",
	})
	defer writer.Close()

	fmt.Println("Inventory Service listening for payment events...")

	for {
		msg, err := reader.ReadMessage(context.Background())
		if err != nil {
			log.Fatalf("Error reading message: %v", err)
		}

		var event PaymentProcessedEvent
		if err := json.Unmarshal(msg.Value, &event); err != nil {
			log.Printf("Invalid message payload: %v", err)
			continue
		}

		if event.Status == "SUCCESS" {
			log.Printf("[Choreography] Reserved stock for Order: %s", event.OrderID)
			
			// Publish downstream domain event reactively
			resEvent := InventoryReservedEvent{
				OrderID: event.OrderID,
				Status:  "RESERVED",
			}
			payload, _ := json.Marshal(resEvent)
			
			err = writer.WriteMessages(context.Background(), kafka.Message{
				Key:   []byte(event.OrderID),
				Value: payload,
			})
			if err != nil {
				log.Printf("Failed to publish inventory event: %v", err)
			}
		}
	}
}

package main

import (
	"context"
	"errors"
	"fmt"
	"log"
)

type OrderSagaOrchestrator struct {
	paymentClient *PaymentClient
	stockClient   *StockClient
}

func NewOrderSagaOrchestrator(p *PaymentClient, s *StockClient) *OrderSagaOrchestrator {
	return &OrderSagaOrchestrator{paymentClient: p, stockClient: s}
}

func (o *OrderSagaOrchestrator) ExecuteSaga(ctx context.Context, orderID string, amount float64) error {
	log.Printf("[Orchestrator] Starting Saga execution for Order ID: %s", orderID)

	// Step 1: Charge Payment
	if err := o.paymentClient.Charge(ctx, orderID, amount); err != nil {
		log.Printf("[Orchestrator] Payment failed for Order %s: %v", orderID, err)
		return err
	}
	log.Printf("[Orchestrator] Step 1 Complete: Payment Charged")

	// Step 2: Reserve Inventory
	if err := o.stockClient.Reserve(ctx, orderID); err != nil {
		log.Printf("[Orchestrator] Inventory reservation failed: %v. Initiating Compensation...", err)
		
		// Compensation Step: Refund Payment
		if refundErr := o.paymentClient.Refund(ctx, orderID, amount); refundErr != nil {
			log.Printf("[CRITICAL] Compensation failed! Manual intervention required for Order %s", orderID)
		}
		return errors.New("saga aborted: inventory unavailable")
	}

	log.Printf("[Orchestrator] Saga Completed Successfully for Order ID: %s", orderID)
	return nil
}

type PaymentClient struct{}
func (p *PaymentClient) Charge(ctx context.Context, id string, amt float64) error { return nil }
func (p *PaymentClient) Refund(ctx context.Context, id string, amt float64) error { return nil }

type StockClient struct{}
func (s *StockClient) Reserve(ctx context.Context, id string) error { return errors.New("out of stock") }

func main() {
	saga := NewOrderSagaOrchestrator(&PaymentClient{}, &StockClient{})
	_ = saga.ExecuteSaga(context.Background(), "ORD-9982", 149.99)
}

package com.ghaznix.microservices.choreography;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.function.Function;

public record PaymentProcessedEvent(String orderId, String status, double amount) {}
public record InventoryReservedEvent(String orderId, String status) {}

@Configuration
public class InventoryChoreographyProcessor {

    @Bean
    public Function<PaymentProcessedEvent, InventoryReservedEvent> processPaymentEvent() {
        return paymentEvent -> {
            System.out.println("[Choreography Java] Processing payment event for order: " + paymentEvent.orderId());

            if ("SUCCESS".equals(paymentEvent.status())) {
                // Reserve stock in database...
                System.out.println("[Choreography Java] Reserved inventory for: " + paymentEvent.orderId());
                return new InventoryReservedEvent(paymentEvent.orderId(), "SUCCESS");
            } else {
                return new InventoryReservedEvent(paymentEvent.orderId(), "FAILED");
            }
        };
    }
}

package com.ghaznix.microservices.orchestration;

import org.springframework.stereotype.Service;

@Service
public class OrderSagaOrchestratorService {

    private final PaymentServiceClient paymentClient;
    private final InventoryServiceClient inventoryClient;
    private final ShippingServiceClient shippingClient;

    public OrderSagaOrchestratorService(PaymentServiceClient p, InventoryServiceClient i, ShippingServiceClient s) {
        this.paymentClient = p;
        this.inventoryClient = i;
        this.shippingClient = s;
    }

    public boolean processCheckoutSaga(String orderId, double totalAmount) {
        System.out.println("[Orchestrator Java] Initiating Saga Workflow for Order: " + orderId);

        // Step 1: Execute Payment
        boolean paymentSuccess = paymentClient.processPayment(orderId, totalAmount);
        if (!paymentSuccess) {
            System.err.println("[Orchestrator Java] Step 1 Failed: Aborting Saga.");
            return false;
        }

        // Step 2: Reserve Inventory
        boolean inventorySuccess = inventoryClient.reserveStock(orderId);
        if (!inventorySuccess) {
            System.err.println("[Orchestrator Java] Step 2 Failed: Triggering Compensation.");
            paymentClient.refundPayment(orderId, totalAmount);
            return false;
        }

        // Step 3: Trigger Shipping
        boolean shippingSuccess = shippingClient.createShipment(orderId);
        if (!shippingSuccess) {
            System.err.println("[Orchestrator Java] Step 3 Failed: Compensating Step 2 & Step 1.");
            inventoryClient.releaseStock(orderId);
            paymentClient.refundPayment(orderId, totalAmount);
            return false;
        }

        System.out.println("[Orchestrator Java] Saga Executed Successfully.");
        return true;
    }
}