Очереди сообщений: сравнение и выбор

<?php

declare(strict_types=1);

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Exchange\AMQPExchangeType;

/**
 * RabbitMQ DLQ configuration example
 */
final class RabbitMqDlqSetup
{
    public function configure(AMQPStreamConnection $connection): void
    {
        $channel = $connection->channel();

        // Declare DLQ exchange and queue
        $channel->exchange_declare(
            exchange: 'dlx.exchange',
            type: AMQPExchangeType::DIRECT,
            durable: true,
        );

        $channel->queue_declare(
            queue: 'orders.dlq',
            durable: true,
        );

        $channel->queue_bind(
            queue: 'orders.dlq',
            exchange: 'dlx.exchange',
            routing_key: 'orders',
        );

        // Declare main queue with DLQ settings
        $channel->queue_declare(
            queue: 'orders',
            durable: true,
            arguments: [
                'x-dead-letter-exchange' => ['S', 'dlx.exchange'],
                'x-dead-letter-routing-key' => ['S', 'orders'],
                'x-message-ttl' => ['I', 60000],        // 60s TTL
                'x-max-delivery-count' => ['I', 3],      // max 3 attempts (RabbitMQ 3.10+)
            ],
        );

        $channel->exchange_declare(
            exchange: 'orders.exchange',
            type: AMQPExchangeType::DIRECT,
            durable: true,
        );

        $channel->queue_bind(
            queue: 'orders',
            exchange: 'orders.exchange',
            routing_key: 'orders.created',
        );
    }
}

package queue

import (
	"fmt"

	amqp "github.com/rabbitmq/amqp091-go"
)

// SetupDLQ configures main queue with dead-letter exchange routing.
func SetupDLQ(ch *amqp.Channel) error {
	// Declare DLX exchange
	if err := ch.ExchangeDeclare(
		"dlx.exchange", "direct", true, false, false, false, nil,
	); err != nil {
		return fmt.Errorf("declare dlx exchange: %w", err)
	}

	// Declare DLQ queue
	if _, err := ch.QueueDeclare(
		"orders.dlq", true, false, false, false, nil,
	); err != nil {
		return fmt.Errorf("declare dlq queue: %w", err)
	}

	if err := ch.QueueBind("orders.dlq", "orders", "dlx.exchange", false, nil); err != nil {
		return fmt.Errorf("bind dlq: %w", err)
	}

	// Declare main queue with DLQ routing
	args := amqp.Table{
		"x-dead-letter-exchange":    "dlx.exchange",
		"x-dead-letter-routing-key": "orders",
		"x-message-ttl":             int64(60000),
	}

	if _, err := ch.QueueDeclare(
		"orders", true, false, false, false, args,
	); err != nil {
		return fmt.Errorf("declare main queue: %w", err)
	}

	return nil
}

using RabbitMQ.Client;

namespace App.Queue;

/// RabbitMQ DLQ configuration example.
public sealed class RabbitMqDlqSetup
{
    public async Task ConfigureAsync(IConnection connection, CancellationToken ct = default)
    {
        await using var channel = await connection.CreateChannelAsync(cancellationToken: ct);

        // Declare DLQ exchange and queue
        await channel.ExchangeDeclareAsync(
            exchange: "dlx.exchange",
            type: ExchangeType.Direct,
            durable: true,
            cancellationToken: ct);

        await channel.QueueDeclareAsync(
            queue: "orders.dlq",
            durable: true,
            exclusive: false,
            autoDelete: false,
            cancellationToken: ct);

        await channel.QueueBindAsync(
            queue: "orders.dlq",
            exchange: "dlx.exchange",
            routingKey: "orders",
            cancellationToken: ct);

        // Declare main queue with DLQ settings
        var args = new Dictionary<string, object?>
        {
            ["x-dead-letter-exchange"] = "dlx.exchange",
            ["x-dead-letter-routing-key"] = "orders",
            ["x-message-ttl"] = 60000,          // 60s TTL
            ["x-max-delivery-count"] = 3,        // max 3 attempts (RabbitMQ 3.10+)
        };

        await channel.QueueDeclareAsync(
            queue: "orders",
            durable: true,
            exclusive: false,
            autoDelete: false,
            arguments: args,
            cancellationToken: ct);

        await channel.ExchangeDeclareAsync(
            exchange: "orders.exchange",
            type: ExchangeType.Direct,
            durable: true,
            cancellationToken: ct);

        await channel.QueueBindAsync(
            queue: "orders",
            exchange: "orders.exchange",
            routingKey: "orders.created",
            cancellationToken: ct);
    }
}

import pika

# RabbitMQ DLQ configuration example

def configure(connection: pika.BlockingConnection) -> None:
    channel = connection.channel()

    # Declare DLQ exchange and queue
    channel.exchange_declare(
        exchange="dlx.exchange",
        exchange_type="direct",
        durable=True,
    )

    channel.queue_declare(queue="orders.dlq", durable=True)

    channel.queue_bind(
        queue="orders.dlq",
        exchange="dlx.exchange",
        routing_key="orders",
    )

    # Declare main queue with DLQ settings
    channel.queue_declare(
        queue="orders",
        durable=True,
        arguments={
            "x-dead-letter-exchange": "dlx.exchange",
            "x-dead-letter-routing-key": "orders",
            "x-message-ttl": 60000,        # 60s TTL
            "x-max-delivery-count": 3,      # max 3 attempts (RabbitMQ 3.10+)
        },
    )

    channel.exchange_declare(
        exchange="orders.exchange",
        exchange_type="direct",
        durable=True,
    )

    channel.queue_bind(
        queue="orders",
        exchange="orders.exchange",
        routing_key="orders.created",
    )

<?php

declare(strict_types=1);

/**
 * Circuit breaker prevents cascading failures when downstream is unhealthy.
 */
final class CircuitBreaker
{
    private int $failures = 0;
    private float $lastFailure = 0;

    public function __construct(
        private readonly int $threshold = 5,
        private readonly int $resetTimeoutSeconds = 60,
    ) {}

    public function isOpen(): bool
    {
        if ($this->failures < $this->threshold) {
            return false;
        }

        // After timeout -- allow one probe request (half-open)
        if ((microtime(true) - $this->lastFailure) > $this->resetTimeoutSeconds) {
            return false;
        }

        return true;
    }

    public function recordSuccess(): void
    {
        $this->failures = 0;
    }

    public function recordFailure(): void
    {
        $this->failures++;
        $this->lastFailure = microtime(true);
    }
}

/**
 * Retry budget limits total retry percentage to prevent thundering herd.
 * Rule: retries should not exceed 10% of successful requests.
 */
final class RetryBudget
{
    private int $totalRequests = 0;
    private int $retryCount = 0;

    public function __construct(
        private readonly float $maxRetryRatio = 0.1, // 10%
        private readonly int $minRetriesPerSecond = 3,
    ) {}

    public function canRetry(): bool
    {
        if ($this->totalRequests === 0) {
            return true;
        }

        // Always allow minimum retries
        if ($this->retryCount < $this->minRetriesPerSecond) {
            return true;
        }

        return ($this->retryCount / $this->totalRequests) < $this->maxRetryRatio;
    }

    public function recordRequest(): void
    {
        $this->totalRequests++;
    }

    public function recordRetry(): void
    {
        $this->retryCount++;
    }
}

package retry

import (
	"math"
	"math/rand/v2"
	"sync"
	"time"
)

// ExponentialBackoff calculates delay with jitter for retry attempt.
func ExponentialBackoff(attempt int, baseDelay time.Duration, maxDelay time.Duration) time.Duration {
	delay := baseDelay * time.Duration(math.Pow(2, float64(attempt)))
	if delay > maxDelay {
		delay = maxDelay
	}
	// Add jitter: ±25%
	jitter := time.Duration(rand.Int64N(int64(delay) / 2))
	return delay/2 + jitter
}

// CircuitBreaker prevents cascading failures.
type CircuitBreaker struct {
	mu               sync.Mutex
	failures         int
	threshold        int
	lastFailure      time.Time
	resetTimeout     time.Duration
}

func NewCircuitBreaker(threshold int, resetTimeout time.Duration) *CircuitBreaker {
	return &CircuitBreaker{
		threshold:    threshold,
		resetTimeout: resetTimeout,
	}
}

// IsOpen returns true if circuit is open (reject requests).
func (cb *CircuitBreaker) IsOpen() bool {
	cb.mu.Lock()
	defer cb.mu.Unlock()

	if cb.failures < cb.threshold {
		return false
	}
	// Half-open: allow probe after timeout
	return time.Since(cb.lastFailure) <= cb.resetTimeout
}

func (cb *CircuitBreaker) RecordSuccess() {
	cb.mu.Lock()
	defer cb.mu.Unlock()
	cb.failures = 0
}

func (cb *CircuitBreaker) RecordFailure() {
	cb.mu.Lock()
	defer cb.mu.Unlock()
	cb.failures++
	cb.lastFailure = time.Now()
}

namespace App.Resilience;

/// Circuit breaker prevents cascading failures when downstream is unhealthy.
public sealed class CircuitBreaker(int threshold = 5, TimeSpan? resetTimeout = null)
{
    private readonly TimeSpan _resetTimeout = resetTimeout ?? TimeSpan.FromSeconds(60);
    private readonly Lock _gate = new();
    private int _failures;
    private DateTimeOffset _lastFailure;

    /// Returns true if circuit is open (reject requests).
    public bool IsOpen()
    {
        lock (_gate)
        {
            if (_failures < threshold)
            {
                return false;
            }

            // Half-open: allow probe after timeout
            return DateTimeOffset.UtcNow - _lastFailure <= _resetTimeout;
        }
    }

    public void RecordSuccess()
    {
        lock (_gate)
        {
            _failures = 0;
        }
    }

    public void RecordFailure()
    {
        lock (_gate)
        {
            _failures++;
            _lastFailure = DateTimeOffset.UtcNow;
        }
    }
}

/// Retry budget limits total retry percentage to prevent thundering herd.
/// Rule: retries should not exceed 10% of successful requests.
public sealed class RetryBudget(double maxRetryRatio = 0.1, int minRetriesPerSecond = 3)
{
    private long _totalRequests;
    private long _retryCount;

    public bool CanRetry()
    {
        var total = Interlocked.Read(ref _totalRequests);
        if (total == 0)
        {
            return true;
        }

        var retries = Interlocked.Read(ref _retryCount);

        // Always allow minimum retries
        if (retries < minRetriesPerSecond)
        {
            return true;
        }

        return (double)retries / total < maxRetryRatio;
    }

    public void RecordRequest() => Interlocked.Increment(ref _totalRequests);

    public void RecordRetry() => Interlocked.Increment(ref _retryCount);
}

/// ExponentialBackoff calculates delay with jitter for retry attempt.
public static class Backoff
{
    public static TimeSpan Exponential(int attempt, TimeSpan baseDelay, TimeSpan maxDelay)
    {
        var delay = baseDelay * Math.Pow(2, attempt);
        if (delay > maxDelay)
        {
            delay = maxDelay;
        }

        // Add jitter: ±25%
        var jitter = Random.Shared.NextDouble() * delay.TotalMilliseconds / 2;
        return TimeSpan.FromMilliseconds(delay.TotalMilliseconds / 2 + jitter);
    }
}

import random
import threading
import time

class CircuitBreaker:
    """Circuit breaker prevents cascading failures when downstream is unhealthy."""

    def __init__(self, threshold: int = 5, reset_timeout_seconds: float = 60.0) -> None:
        self._threshold = threshold
        self._reset_timeout = reset_timeout_seconds
        self._lock = threading.Lock()
        self._failures = 0
        self._last_failure = 0.0

    def is_open(self) -> bool:
        with self._lock:
            if self._failures < self._threshold:
                return False

            # After timeout -- allow one probe request (half-open)
            return (time.monotonic() - self._last_failure) <= self._reset_timeout

    def record_success(self) -> None:
        with self._lock:
            self._failures = 0

    def record_failure(self) -> None:
        with self._lock:
            self._failures += 1
            self._last_failure = time.monotonic()

class RetryBudget:
    """Retry budget limits total retry percentage to prevent thundering herd.

    Rule: retries should not exceed 10% of successful requests.
    """

    def __init__(self, max_retry_ratio: float = 0.1, min_retries_per_second: int = 3) -> None:
        self._max_retry_ratio = max_retry_ratio
        self._min_retries_per_second = min_retries_per_second
        self._lock = threading.Lock()
        self._total_requests = 0
        self._retry_count = 0

    def can_retry(self) -> bool:
        with self._lock:
            if self._total_requests == 0:
                return True

            # Always allow minimum retries
            if self._retry_count < self._min_retries_per_second:
                return True

            return (self._retry_count / self._total_requests) < self._max_retry_ratio

    def record_request(self) -> None:
        with self._lock:
            self._total_requests += 1

    def record_retry(self) -> None:
        with self._lock:
            self._retry_count += 1

def exponential_backoff(attempt: int, base_delay: float, max_delay: float) -> float:
    """Calculate delay in seconds with jitter for retry attempt."""
    delay = min(base_delay * (2 ** attempt), max_delay)
    # Add jitter: ±25%
    return delay / 2 + random.uniform(0, delay / 2)

<?php

declare(strict_types=1);

namespace App\Message;

/**
 * Dispatched when a new order is placed.
 * Immutable message — all data set at construction.
 */
final readonly class OrderCreated
{
    public function __construct(
        public string $orderId,
        public string $userId,
        public float $totalAmount,
        public string $currency = 'USD',
        public \DateTimeImmutable $occurredAt = new \DateTimeImmutable(),
    ) {}
}

<?php

declare(strict_types=1);

namespace App\MessageHandler;

use App\Message\OrderCreated;
use App\Service\InventoryService;
use App\Service\InvoiceService;
use Psr\Log\LoggerInterface;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;

#[AsMessageHandler]
final readonly class OrderCreatedHandler
{
    public function __construct(
        private InventoryService $inventory,
        private InvoiceService $invoice,
        private LoggerInterface $logger,
    ) {}

    /**
     * Process new order: reserve inventory, generate invoice.
     * Must be idempotent — message may be delivered more than once.
     */
    public function __invoke(OrderCreated $message): void
    {
        $this->logger->info('Processing order', [
            'orderId' => $message->orderId,
            'userId' => $message->userId,
        ]);

        // Idempotent: reserve checks if already reserved
        $this->inventory->reserveForOrder($message->orderId);

        // Idempotent: invoice checks if already generated
        $this->invoice->generateForOrder(
            orderId: $message->orderId,
            amount: $message->totalAmount,
            currency: $message->currency,
        );

        $this->logger->info('Order processed successfully', [
            'orderId' => $message->orderId,
        ]);
    }
}

<?php

declare(strict_types=1);

namespace App\Controller;

use App\Message\OrderCreated;
use Symfony\Component\HttpFoundation\JsonResponse;
use Symfony\Component\HttpFoundation\Request;
use Symfony\Component\HttpFoundation\Response;
use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Routing\Attribute\Route;

final readonly class OrderController
{
    public function __construct(
        private MessageBusInterface $bus,
    ) {}

    #[Route('/api/v1/orders', methods: ['POST'])]
    public function create(Request $request): JsonResponse
    {
        $data = json_decode($request->getContent(), true, flags: JSON_THROW_ON_ERROR);

        // Save order to DB synchronously
        $orderId = $this->persistOrder($data);

        // Dispatch async processing
        $this->bus->dispatch(new OrderCreated(
            orderId: $orderId,
            userId: $data['userId'],
            totalAmount: (float) $data['totalAmount'],
            currency: $data['currency'] ?? 'USD',
        ));

        // Instant response — processing happens asynchronously
        return new JsonResponse(
            data: ['orderId' => $orderId, 'status' => 'accepted'],
            status: Response::HTTP_ACCEPTED,
        );
    }

    private function persistOrder(array $data): string
    {
        // ... save to database, return generated ID
        return 'ord_' . bin2hex(random_bytes(8));
    }
}

package consumer

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

	amqp "github.com/rabbitmq/amqp091-go"
)

// OrderCreated represents an incoming order event.
type OrderCreated struct {
	OrderID     string    `json:"orderId"`
	UserID      string    `json:"userId"`
	TotalAmount float64   `json:"totalAmount"`
	Currency    string    `json:"currency"`
	OccurredAt  time.Time `json:"occurredAt"`
}

// RabbitConsumer processes messages from RabbitMQ with manual acknowledgment.
type RabbitConsumer struct {
	conn    *amqp.Connection
	channel *amqp.Channel
	logger  *slog.Logger
}

func NewRabbitConsumer(url string, logger *slog.Logger) (*RabbitConsumer, error) {
	conn, err := amqp.Dial(url)
	if err != nil {
		return nil, fmt.Errorf("dial rabbitmq: %w", err)
	}

	ch, err := conn.Channel()
	if err != nil {
		return nil, fmt.Errorf("open channel: %w", err)
	}

	// Prefetch: process one message at a time
	if err := ch.Qos(1, 0, false); err != nil {
		return nil, fmt.Errorf("set qos: %w", err)
	}

	return &RabbitConsumer{conn: conn, channel: ch, logger: logger}, nil
}

// Consume starts consuming messages from the given queue.
// Blocks until context is cancelled.
func (c *RabbitConsumer) Consume(ctx context.Context, queue string, handler func(OrderCreated) error) error {
	msgs, err := c.channel.Consume(
		queue,
		"",    // consumer tag (auto-generated)
		false, // auto-ack: false = manual ACK
		false, // exclusive
		false, // no-local
		false, // no-wait
		nil,
	)
	if err != nil {
		return fmt.Errorf("consume: %w", err)
	}

	c.logger.Info("consumer started", "queue", queue)

	for {
		select {
		case <-ctx.Done():
			c.logger.Info("consumer stopping", "reason", ctx.Err())
			return ctx.Err()

		case msg, ok := <-msgs:
			if !ok {
				return fmt.Errorf("channel closed")
			}

			var order OrderCreated
			if err := json.Unmarshal(msg.Body, &order); err != nil {
				c.logger.Error("invalid message payload", "error", err)
				// Reject without requeue — goes to DLQ
				_ = msg.Nack(false, false)
				continue
			}

			if err := handler(order); err != nil {
				c.logger.Error("handler failed",
					"orderId", order.OrderID,
					"error", err,
				)
				// Nack with requeue: broker will redeliver
				_ = msg.Nack(false, true)
				continue
			}

			// Successful processing — acknowledge
			if err := msg.Ack(false); err != nil {
				c.logger.Error("ack failed", "error", err)
			}

			c.logger.Info("order processed",
				"orderId", order.OrderID,
				"amount", order.TotalAmount,
			)
		}
	}
}

func (c *RabbitConsumer) Close() error {
	if err := c.channel.Close(); err != nil {
		return fmt.Errorf("close channel: %w", err)
	}
	return c.conn.Close()
}

package main

import (
	"context"
	"fmt"
	"log/slog"
	"os"
	"os/signal"
	"syscall"

	"myapp/consumer"
)

func main() {
	logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
		Level: slog.LevelInfo,
	}))

	c, err := consumer.NewRabbitConsumer(
		"amqp://guest:guest@localhost:5672/",
		logger,
	)
	if err != nil {
		logger.Error("failed to create consumer", "error", err)
		os.Exit(1)
	}
	defer c.Close()

	ctx, cancel := signal.NotifyContext(context.Background(),
		syscall.SIGINT, syscall.SIGTERM,
	)
	defer cancel()

	// Handler: business logic for processing orders
	handler := func(order consumer.OrderCreated) error {
		logger.Info("handling order",
			"orderId", order.OrderID,
			"user", order.UserID,
		)

		// Idempotent processing: check if already processed
		// ... reserve inventory, generate invoice, etc.

		return nil
	}

	if err := c.Consume(ctx, "orders", handler); err != nil {
		fmt.Fprintf(os.Stderr, "consumer error: %v\n", err)
	}
}

package consumer

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

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

// KafkaConsumer reads messages from Kafka topic using consumer group.
type KafkaConsumer struct {
	reader *kafka.Reader
	logger *slog.Logger
}

// NewKafkaConsumer creates a consumer with consumer group for parallel processing.
func NewKafkaConsumer(brokers []string, topic, groupID string, logger *slog.Logger) *KafkaConsumer {
	reader := kafka.NewReader(kafka.ReaderConfig{
		Brokers:        brokers,
		Topic:          topic,
		GroupID:        groupID,
		MinBytes:       1,              // 1 byte
		MaxBytes:       10 * 1024 * 1024, // 10 MB
		MaxWait:        500 * time.Millisecond,
		CommitInterval: time.Second,     // Auto-commit interval
		StartOffset:    kafka.FirstOffset,
	})

	return &KafkaConsumer{reader: reader, logger: logger}
}

// Event is a generic event read from Kafka.
type Event struct {
	Key       string
	Value     json.RawMessage
	Topic     string
	Partition int
	Offset    int64
	Timestamp time.Time
}

// Consume reads messages until context is cancelled.
// Handler receives deserialized events. Offset is committed
// only after successful processing (at-least-once).
func (c *KafkaConsumer) Consume(ctx context.Context, handler func(ctx context.Context, event Event) error) error {
	c.logger.Info("kafka consumer started",
		"topic", c.reader.Config().Topic,
		"group", c.reader.Config().GroupID,
	)

	for {
		// FetchMessage does NOT auto-commit offset
		msg, err := c.reader.FetchMessage(ctx)
		if err != nil {
			if ctx.Err() != nil {
				return ctx.Err()
			}
			c.logger.Error("fetch failed", "error", err)
			continue
		}

		event := Event{
			Key:       string(msg.Key),
			Value:     msg.Value,
			Topic:     msg.Topic,
			Partition: msg.Partition,
			Offset:    msg.Offset,
			Timestamp: msg.Time,
		}

		c.logger.Info("message received",
			"partition", msg.Partition,
			"offset", msg.Offset,
			"key", string(msg.Key),
		)

		// Process message
		if err := handler(ctx, event); err != nil {
			c.logger.Error("handler error",
				"partition", msg.Partition,
				"offset", msg.Offset,
				"error", err,
			)
			// Do not commit — message will be redelivered on restart
			continue
		}

		// Commit offset only after successful processing
		if err := c.reader.CommitMessages(ctx, msg); err != nil {
			c.logger.Error("commit failed", "error", err)
		}
	}
}

func (c *KafkaConsumer) Close() error {
	return c.reader.Close()
}

// Producer demonstrates sending messages with partition key.
type Producer struct {
	writer *kafka.Writer
}

func NewProducer(brokers []string, topic string) *Producer {
	return &Producer{
		writer: &kafka.Writer{
			Addr:     kafka.TCP(brokers...),
			Topic:    topic,
			Balancer: &kafka.Hash{}, // Same key → same partition
		},
	}
}

// Send publishes a message with partition key for ordering guarantee.
func (p *Producer) Send(ctx context.Context, key string, payload any) error {
	data, err := json.Marshal(payload)
	if err != nil {
		return fmt.Errorf("marshal: %w", err)
	}

	return p.writer.WriteMessages(ctx, kafka.Message{
		Key:   []byte(key),   // Partition key: all events for this key go to same partition
		Value: data,
		Time:  time.Now(),
	})
}

func (p *Producer) Close() error {
	return p.writer.Close()
}

using System.Text.Json;
using Confluent.Kafka;
using Microsoft.Extensions.Logging;

namespace App.Kafka;

/// Event is a generic event read from Kafka.
public sealed record Event(
    string Key,
    JsonElement Value,
    string Topic,
    int Partition,
    long Offset,
    DateTime Timestamp);

/// KafkaConsumer reads messages from a Kafka topic using a consumer group.
public sealed class KafkaConsumer : IDisposable
{
    private readonly IConsumer<string, string> _consumer;
    private readonly ILogger _logger;
    private readonly string _topic;

    public KafkaConsumer(string brokers, string topic, string groupId, ILogger logger)
    {
        var config = new ConsumerConfig
        {
            BootstrapServers = brokers,
            GroupId = groupId,
            AutoOffsetReset = AutoOffsetReset.Earliest,
            EnableAutoCommit = false,       // Commit manually after processing
            FetchWaitMaxMs = 500,
            FetchMaxBytes = 10 * 1024 * 1024, // 10 MB
        };

        _consumer = new ConsumerBuilder<string, string>(config).Build();
        _logger = logger;
        _topic = topic;
    }

    /// Consume reads messages until the token is cancelled.
    /// Offset is committed only after successful processing (at-least-once).
    public async Task ConsumeAsync(Func<Event, CancellationToken, Task> handler, CancellationToken ct)
    {
        _consumer.Subscribe(_topic);
        _logger.LogInformation("kafka consumer started, topic {Topic}", _topic);

        while (!ct.IsCancellationRequested)
        {
            ConsumeResult<string, string> result;
            try
            {
                // Consume does NOT auto-commit offset (EnableAutoCommit = false)
                result = _consumer.Consume(ct);
            }
            catch (ConsumeException ex)
            {
                _logger.LogError(ex, "fetch failed");
                continue;
            }
            catch (OperationCanceledException)
            {
                break;
            }

            var evt = new Event(
                result.Message.Key,
                JsonDocument.Parse(result.Message.Value).RootElement,
                result.Topic,
                result.Partition.Value,
                result.Offset.Value,
                result.Message.Timestamp.UtcDateTime);

            _logger.LogInformation("message received partition {Partition} offset {Offset} key {Key}",
                evt.Partition, evt.Offset, evt.Key);

            try
            {
                await handler(evt, ct);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "handler error partition {Partition} offset {Offset}",
                    evt.Partition, evt.Offset);
                // Do not commit — message will be redelivered on restart
                continue;
            }

            // Commit offset only after successful processing
            _consumer.Commit(result);
        }

        _consumer.Close();
    }

    public void Dispose() => _consumer.Dispose();
}

/// Producer demonstrates sending messages with a partition key.
public sealed class Producer(string brokers, string topic) : IDisposable
{
    private readonly IProducer<string, string> _producer =
        new ProducerBuilder<string, string>(new ProducerConfig { BootstrapServers = brokers }).Build();

    /// Send publishes a message with partition key for ordering guarantee.
    public async Task SendAsync(string key, object payload)
    {
        var message = new Message<string, string>
        {
            // Partition key: all events for this key go to same partition
            Key = key,
            Value = JsonSerializer.Serialize(payload),
            Timestamp = new Timestamp(DateTime.UtcNow),
        };

        await _producer.ProduceAsync(topic, message);
    }

    public void Dispose()
    {
        _producer.Flush(TimeSpan.FromSeconds(10));
        _producer.Dispose();
    }
}

import json
import logging
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Awaitable, Callable

from confluent_kafka import Consumer, KafkaError, KafkaException, Producer

logger = logging.getLogger(__name__)

@dataclass(frozen=True, slots=True)
class Event:
    """Event is a generic event read from Kafka."""

    key: str
    value: Any
    topic: str
    partition: int
    offset: int
    timestamp: datetime

class KafkaConsumer:
    """Reads messages from a Kafka topic using a consumer group."""

    def __init__(self, brokers: list[str], topic: str, group_id: str) -> None:
        self._consumer = Consumer({
            "bootstrap.servers": ",".join(brokers),
            "group.id": group_id,
            "auto.offset.reset": "earliest",
            "enable.auto.commit": False,   # Commit manually after processing
            "fetch.wait.max.ms": 500,
            "fetch.max.bytes": 10 * 1024 * 1024,  # 10 MB
        })
        self._topic = topic

    def consume(self, handler: Callable[[Event], None]) -> None:
        """Read messages until interrupted.

        Offset is committed only after successful processing (at-least-once).
        """
        self._consumer.subscribe([self._topic])
        logger.info("kafka consumer started", extra={"topic": self._topic})

        while True:
            # poll() does NOT auto-commit offset (enable.auto.commit=False)
            msg = self._consumer.poll(timeout=1.0)
            if msg is None:
                continue

            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    continue
                logger.error("fetch failed", extra={"error": str(msg.error())})
                continue

            event = Event(
                key=msg.key().decode() if msg.key() else "",
                value=json.loads(msg.value()),
                topic=msg.topic(),
                partition=msg.partition(),
                offset=msg.offset(),
                timestamp=datetime.fromtimestamp(msg.timestamp()[1] / 1000, tz=timezone.utc),
            )

            logger.info("message received", extra={
                "partition": event.partition,
                "offset": event.offset,
                "key": event.key,
            })

            try:
                handler(event)
            except Exception as exc:
                logger.error("handler error", extra={
                    "partition": event.partition,
                    "offset": event.offset,
                    "error": str(exc),
                })
                # Do not commit — message will be redelivered on restart
                continue

            # Commit offset only after successful processing
            self._consumer.commit(msg, asynchronous=False)

    def close(self) -> None:
        self._consumer.close()

class KafkaProducer:
    """Demonstrates sending messages with a partition key."""

    def __init__(self, brokers: list[str], topic: str) -> None:
        # Default partitioner hashes the key: same key → same partition
        self._producer = Producer({"bootstrap.servers": ",".join(brokers)})
        self._topic = topic

    def send(self, key: str, payload: Any) -> None:
        """Publish a message with partition key for ordering guarantee."""
        self._producer.produce(
            self._topic,
            key=key.encode(),
            value=json.dumps(payload).encode(),
        )
        self._producer.poll(0)

    def close(self) -> None:
        # Block until all queued messages are delivered
        self._producer.flush(10)

<?php

declare(strict_types=1);

namespace App\Service;

use Doctrine\DBAL\Connection;

/**
 * Idempotent message processor using deduplication table.
 * Prevents double-processing on redelivery.
 */
final readonly class IdempotentProcessor
{
    public function __construct(
        private Connection $db,
    ) {}

    /**
     * Process message only if not already processed.
     * Uses DB transaction: dedup check + business logic are atomic.
     *
     * @param string $messageId Unique message identifier
     * @param callable $handler Business logic to execute
     */
    public function processOnce(string $messageId, callable $handler): void
    {
        $this->db->transactional(function () use ($messageId, $handler): void {
            // Check if already processed (SELECT FOR UPDATE prevents race)
            $exists = $this->db->fetchOne(
                'SELECT 1 FROM processed_messages WHERE message_id = ? FOR UPDATE',
                [$messageId],
            );

            if ($exists !== false) {
                return; // Already processed — skip
            }

            // Execute business logic
            $handler();

            // Mark as processed
            $this->db->executeStatement(
                'INSERT INTO processed_messages (message_id, processed_at) VALUES (?, NOW())',
                [$messageId],
            );
        });
    }
}

// Usage in handler:
// $this->idempotentProcessor->processOnce(
//     $message->orderId,
//     fn() => $this->processOrder($message),
// );

package dedup

import (
	"context"
	"database/sql"
	"fmt"
)

// IdempotentProcessor ensures each message is processed exactly once
// using a deduplication table in PostgreSQL.
type IdempotentProcessor struct {
	db *sql.DB
}

func NewIdempotentProcessor(db *sql.DB) *IdempotentProcessor {
	return &IdempotentProcessor{db: db}
}

// ProcessOnce executes handler only if messageID has not been processed before.
// Dedup check and handler execute in a single transaction.
func (p *IdempotentProcessor) ProcessOnce(ctx context.Context, messageID string, handler func(tx *sql.Tx) error) error {
	tx, err := p.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin tx: %w", err)
	}
	defer tx.Rollback()

	// Check with row-level lock
	var exists bool
	err = tx.QueryRowContext(ctx,
		"SELECT EXISTS(SELECT 1 FROM processed_messages WHERE message_id = $1 FOR UPDATE)",
		messageID,
	).Scan(&exists)
	if err != nil {
		return fmt.Errorf("check dedup: %w", err)
	}

	if exists {
		return nil // Already processed
	}

	// Execute business logic inside same transaction
	if err := handler(tx); err != nil {
		return fmt.Errorf("handler: %w", err)
	}

	// Mark as processed
	if _, err := tx.ExecContext(ctx,
		"INSERT INTO processed_messages (message_id, processed_at) VALUES ($1, NOW())",
		messageID,
	); err != nil {
		return fmt.Errorf("insert dedup: %w", err)
	}

	return tx.Commit()
}

using Npgsql;

namespace App.Dedup;

/// Idempotent message processor using a deduplication table.
/// Prevents double-processing on redelivery.
public sealed class IdempotentProcessor(NpgsqlDataSource dataSource)
{
    /// Executes handler only if messageId has not been processed before.
    /// Dedup check and handler run in a single transaction.
    public async Task ProcessOnceAsync(
        string messageId,
        Func<NpgsqlConnection, NpgsqlTransaction, CancellationToken, Task> handler,
        CancellationToken ct = default)
    {
        await using var connection = await dataSource.OpenConnectionAsync(ct);
        await using var tx = await connection.BeginTransactionAsync(ct);

        // Check with row-level lock
        await using var check = new NpgsqlCommand(
            "SELECT 1 FROM processed_messages WHERE message_id = @id FOR UPDATE",
            connection, tx);
        check.Parameters.AddWithValue("id", messageId);

        if (await check.ExecuteScalarAsync(ct) is not null)
        {
            return; // Already processed — skip
        }

        // Execute business logic inside same transaction
        await handler(connection, tx, ct);

        // Mark as processed
        await using var mark = new NpgsqlCommand(
            "INSERT INTO processed_messages (message_id, processed_at) VALUES (@id, NOW())",
            connection, tx);
        mark.Parameters.AddWithValue("id", messageId);
        await mark.ExecuteNonQueryAsync(ct);

        await tx.CommitAsync(ct);
    }
}

// Usage in consumer:
// await _idempotentProcessor.ProcessOnceAsync(
//     message.OrderId,
//     (conn, tx, ct) => ProcessOrderAsync(message, conn, tx, ct),
//     cancellationToken);

from typing import Awaitable, Callable

import asyncpg

class IdempotentProcessor:
    """Idempotent message processor using a deduplication table.

    Prevents double-processing on redelivery.
    """

    def __init__(self, pool: asyncpg.Pool) -> None:
        self._pool = pool

    async def process_once(
        self,
        message_id: str,
        handler: Callable[[asyncpg.Connection], Awaitable[None]],
    ) -> None:
        """Execute handler only if message_id has not been processed before.

        Dedup check and handler run in a single transaction.
        """
        async with self._pool.acquire() as conn:
            async with conn.transaction():
                # Check with row-level lock
                exists = await conn.fetchval(
                    "SELECT 1 FROM processed_messages WHERE message_id = $1 FOR UPDATE",
                    message_id,
                )

                if exists is not None:
                    return  # Already processed — skip

                # Execute business logic inside same transaction
                await handler(conn)

                # Mark as processed
                await conn.execute(
                    "INSERT INTO processed_messages (message_id, processed_at) VALUES ($1, NOW())",
                    message_id,
                )

# Usage in handler:
# await idempotent_processor.process_once(
#     message.order_id,
#     lambda conn: process_order(message, conn),
# )

<?php

declare(strict_types=1);

namespace App\Service;

use App\Entity\Order;
use App\Entity\OutboxMessage;
use Doctrine\ORM\EntityManagerInterface;

/**
 * Transactional outbox: business write + message in single DB transaction.
 * A relay process (cron/daemon) publishes outbox messages to the broker.
 */
final readonly class OrderService
{
    public function __construct(
        private EntityManagerInterface $em,
    ) {}

    public function createOrder(string $userId, float $amount, string $currency): Order
    {
        return $this->em->wrapInTransaction(function () use ($userId, $amount, $currency): Order {
            // Business logic: create order
            $order = new Order(
                userId: $userId,
                amount: $amount,
                currency: $currency,
            );
            $this->em->persist($order);

            // Outbox: store message in same transaction
            $outbox = new OutboxMessage(
                aggregateType: 'order',
                aggregateId: $order->getId(),
                eventType: 'order.created',
                payload: json_encode([
                    'orderId' => $order->getId(),
                    'userId' => $userId,
                    'amount' => $amount,
                    'currency' => $currency,
                ], JSON_THROW_ON_ERROR),
            );
            $this->em->persist($outbox);

            return $order;
        });
    }
}

/**
 * Relay: reads unpublished outbox messages and sends to broker.
 * Runs as a daemon or cron job.
 */
final readonly class OutboxRelay
{
    public function __construct(
        private EntityManagerInterface $em,
        private \Symfony\Component\Messenger\MessageBusInterface $bus,
    ) {}

    public function publishPending(int $batchSize = 100): int
    {
        $messages = $this->em->getRepository(OutboxMessage::class)
            ->findBy(['published' => false], ['createdAt' => 'ASC'], $batchSize);

        $count = 0;
        foreach ($messages as $outbox) {
            $this->bus->dispatch(new \App\Message\OutboxEvent(
                eventType: $outbox->getEventType(),
                payload: $outbox->getPayload(),
            ));

            $outbox->markPublished();
            $count++;
        }

        $this->em->flush();

        return $count;
    }
}

package outbox

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"time"
)

// OutboxMessage represents a message stored in the outbox table.
type OutboxMessage struct {
	ID            string          `json:"id"`
	AggregateType string          `json:"aggregateType"`
	AggregateID   string          `json:"aggregateId"`
	EventType     string          `json:"eventType"`
	Payload       json.RawMessage `json:"payload"`
	Published     bool            `json:"published"`
	CreatedAt     time.Time       `json:"createdAt"`
}

// Store saves business entity and outbox message in a single transaction.
func Store(ctx context.Context, db *sql.DB, orderID, userID string, amount float64) error {
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin tx: %w", err)
	}
	defer tx.Rollback()

	// Business write
	if _, err := tx.ExecContext(ctx,
		"INSERT INTO orders (id, user_id, amount, status) VALUES ($1, $2, $3, 'new')",
		orderID, userID, amount,
	); err != nil {
		return fmt.Errorf("insert order: %w", err)
	}

	// Outbox message in same transaction
	payload, _ := json.Marshal(map[string]any{
		"orderId": orderID,
		"userId":  userID,
		"amount":  amount,
	})

	if _, err := tx.ExecContext(ctx,
		`INSERT INTO outbox (aggregate_type, aggregate_id, event_type, payload, published)
		 VALUES ('order', $1, 'order.created', $2, false)`,
		orderID, payload,
	); err != nil {
		return fmt.Errorf("insert outbox: %w", err)
	}

	return tx.Commit()
}

// Relay reads unpublished messages and sends them to the broker.
// Marks as published after successful send.
type Relay struct {
	db        *sql.DB
	publisher func(ctx context.Context, eventType string, payload []byte) error
}

func (r *Relay) PublishPending(ctx context.Context, batchSize int) (int, error) {
	rows, err := r.db.QueryContext(ctx,
		`SELECT id, event_type, payload FROM outbox
		 WHERE published = false
		 ORDER BY created_at ASC
		 LIMIT $1
		 FOR UPDATE SKIP LOCKED`,
		batchSize,
	)
	if err != nil {
		return 0, fmt.Errorf("query outbox: %w", err)
	}
	defer rows.Close()

	var count int
	for rows.Next() {
		var id, eventType string
		var payload []byte
		if err := rows.Scan(&id, &eventType, &payload); err != nil {
			return count, fmt.Errorf("scan: %w", err)
		}

		if err := r.publisher(ctx, eventType, payload); err != nil {
			return count, fmt.Errorf("publish %s: %w", id, err)
		}

		if _, err := r.db.ExecContext(ctx,
			"UPDATE outbox SET published = true, published_at = NOW() WHERE id = $1", id,
		); err != nil {
			return count, fmt.Errorf("mark published: %w", err)
		}
		count++
	}

	return count, rows.Err()
}

using System.Text.Json;
using Npgsql;

namespace App.Outbox;

/// OutboxMessage represents a message stored in the outbox table.
public sealed record OutboxMessage(
    string Id,
    string AggregateType,
    string AggregateId,
    string EventType,
    JsonElement Payload,
    bool Published,
    DateTimeOffset CreatedAt);

/// Transactional outbox: business write + message in a single DB transaction.
/// A relay process publishes outbox messages to the broker.
public sealed class OrderService(NpgsqlDataSource dataSource)
{
    public async Task StoreAsync(string orderId, string userId, decimal amount, CancellationToken ct)
    {
        await using var connection = await dataSource.OpenConnectionAsync(ct);
        await using var tx = await connection.BeginTransactionAsync(ct);

        // Business write
        await using (var insertOrder = new NpgsqlCommand(
            "INSERT INTO orders (id, user_id, amount, status) VALUES (@id, @user, @amount, 'new')",
            connection, tx))
        {
            insertOrder.Parameters.AddWithValue("id", orderId);
            insertOrder.Parameters.AddWithValue("user", userId);
            insertOrder.Parameters.AddWithValue("amount", amount);
            await insertOrder.ExecuteNonQueryAsync(ct);
        }

        // Outbox message in same transaction
        var payload = JsonSerializer.Serialize(new
        {
            orderId,
            userId,
            amount,
        });

        await using (var insertOutbox = new NpgsqlCommand(
            """
            INSERT INTO outbox (aggregate_type, aggregate_id, event_type, payload, published)
            VALUES ('order', @id, 'order.created', @payload::jsonb, false)
            """,
            connection, tx))
        {
            insertOutbox.Parameters.AddWithValue("id", orderId);
            insertOutbox.Parameters.AddWithValue("payload", payload);
            await insertOutbox.ExecuteNonQueryAsync(ct);
        }

        await tx.CommitAsync(ct);
    }
}

/// Relay reads unpublished messages and sends them to the broker.
/// Marks as published after successful send. Runs as a daemon or cron job.
public sealed class OutboxRelay(
    NpgsqlDataSource dataSource,
    Func<string, string, CancellationToken, Task> publisher)
{
    public async Task<int> PublishPendingAsync(int batchSize, CancellationToken ct)
    {
        await using var connection = await dataSource.OpenConnectionAsync(ct);
        await using var tx = await connection.BeginTransactionAsync(ct);

        var pending = new List<(string Id, string EventType, string Payload)>();

        await using (var select = new NpgsqlCommand(
            """
            SELECT id, event_type, payload FROM outbox
            WHERE published = false
            ORDER BY created_at ASC
            LIMIT @limit
            FOR UPDATE SKIP LOCKED
            """,
            connection, tx))
        {
            select.Parameters.AddWithValue("limit", batchSize);
            await using var reader = await select.ExecuteReaderAsync(ct);
            while (await reader.ReadAsync(ct))
            {
                pending.Add((reader.GetString(0), reader.GetString(1), reader.GetString(2)));
            }
        }

        var count = 0;
        foreach (var (id, eventType, payload) in pending)
        {
            await publisher(eventType, payload, ct);

            await using var update = new NpgsqlCommand(
                "UPDATE outbox SET published = true, published_at = NOW() WHERE id = @id",
                connection, tx);
            update.Parameters.AddWithValue("id", id);
            await update.ExecuteNonQueryAsync(ct);
            count++;
        }

        await tx.CommitAsync(ct);

        return count;
    }
}

import json
from dataclasses import dataclass
from datetime import datetime
from typing import Any, Awaitable, Callable

import asyncpg

@dataclass(frozen=True, slots=True)
class OutboxMessage:
    """OutboxMessage represents a message stored in the outbox table."""

    id: str
    aggregate_type: str
    aggregate_id: str
    event_type: str
    payload: dict[str, Any]
    published: bool
    created_at: datetime

async def store(
    pool: asyncpg.Pool,
    order_id: str,
    user_id: str,
    amount: float,
) -> None:
    """Save business entity and outbox message in a single transaction."""
    async with pool.acquire() as conn:
        async with conn.transaction():
            # Business write
            await conn.execute(
                "INSERT INTO orders (id, user_id, amount, status) VALUES ($1, $2, $3, 'new')",
                order_id, user_id, amount,
            )

            # Outbox message in same transaction
            payload = json.dumps({
                "orderId": order_id,
                "userId": user_id,
                "amount": amount,
            })

            await conn.execute(
                """
                INSERT INTO outbox (aggregate_type, aggregate_id, event_type, payload, published)
                VALUES ('order', $1, 'order.created', $2::jsonb, false)
                """,
                order_id, payload,
            )

class Relay:
    """Reads unpublished messages and sends them to the broker.

    Marks as published after successful send.
    """

    def __init__(
        self,
        pool: asyncpg.Pool,
        publisher: Callable[[str, str], Awaitable[None]],
    ) -> None:
        self._pool = pool
        self._publisher = publisher

    async def publish_pending(self, batch_size: int = 100) -> int:
        async with self._pool.acquire() as conn:
            async with conn.transaction():
                rows = await conn.fetch(
                    """
                    SELECT id, event_type, payload FROM outbox
                    WHERE published = false
                    ORDER BY created_at ASC
                    LIMIT $1
                    FOR UPDATE SKIP LOCKED
                    """,
                    batch_size,
                )

                count = 0
                for row in rows:
                    await self._publisher(row["event_type"], row["payload"])

                    await conn.execute(
                        "UPDATE outbox SET published = true, published_at = NOW() WHERE id = $1",
                        row["id"],
                    )
                    count += 1

                return count