@Configuration
public class RabbitMQConfig {
    
    @Bean
    public CachingConnectionFactory connectionFactory(
            @Value("${spring.rabbitmq.host}") String host) {
        CachingConnectionFactory factory = new CachingConnectionFactory(host);
        
        // Производственные настройки по умолчанию
        factory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        factory.setPublisherReturns(true);
        factory.setChannelCacheSize(25);
        
        return factory;
    }
}

spring:
  rabbitmq:
    ssl:
      enabled: true
      algorithm: TLSv1.3
      key-store: file:/etc/ssl/client-keystore.p12
      key-store-password: ${KEYSTORE_PASSWORD}
      key-store-type: PKCS12
      trust-store: file:/etc/ssl/truststore.jks
      trust-store-password: ${TRUSTSTORE_PASSWORD}
      trust-store-type: JKS
      verify-hostname: true

spring:
  rabbitmq:
    # Подтверждения от брокера
    publisher-confirm-type: correlated
    publisher-returns: true
    
    # Подтверждения от потребителя
    listener:
      simple:
        acknowledge-mode: manual  # Для полного контроля
        prefetch: 10              # Балансировка нагрузки
        retry:
          enabled: true
          max-attempts: 3
          back-off:
            initial-interval: 1s
            multiplier: 2

@Configuration
public class MonitoringConfig {
    
    @Bean
    public MeterRegistryCustomizer<MeterRegistry> rabbitMQMetrics() {
        return registry -> {
            // Кастомные метрики для RabbitMQ
            registry.config().commonTags("application", "order-service");
        };
    }
    
    @Bean
    public RabbitTemplateMonitoring rabbitTemplateMonitoring(
            RabbitTemplate rabbitTemplate,
            MeterRegistry meterRegistry) {
        
        // Мониторинг времени доставки
        Timer timer = Timer.builder("rabbitmq.publish.duration")
            .description("Time to publish message to RabbitMQ")
            .register(meterRegistry);
        
        rabbitTemplate.setObservationEnabled(true);
        
        return new RabbitTemplateMonitoring(rabbitTemplate, timer);
    }
}

@Configuration
@EnableReactiveRabbit
public class ReactiveRabbitConfig {
    
    @Bean
    public ReactiveRabbitTemplate reactiveRabbitTemplate(
            ReactiveRabbitConnectionFactory factory) {
        
        ReactiveRabbitTemplate template = new ReactiveRabbitTemplate(factory);
        
        // Настройки backpressure
        template.setObservationEnabled(true);
        template.setUsePublisherConfirm(true);
        
        return template;
    }
    
    @Bean
    public ReactiveRabbitListenerContainerFactory containerFactory(
            ReactiveRabbitConnectionFactory factory) {
        
        ReactiveRabbitListenerContainerFactory containerFactory = 
            new ReactiveRabbitListenerContainerFactory();
        containerFactory.setConnectionFactory(factory);
        
        // Обработка ошибок в реактивном стиле
        containerFactory.setErrorHandler(errorHandler());
        
        return containerFactory;
    }
    
    private ReactiveErrorHandler errorHandler() {
        return new ReactiveErrorHandler() {
            @Override
            public Mono<Void> handleError(Throwable throwable, 
                                          Acknowledgement acknowledgement) {
                log.error("Reactive consumption error", throwable);
                return acknowledgement.nack(false); // Не requeue при фатальных ошибках
            }
        };
    }
}

@Component
public class OrderEventConsumer {
    
    @ReactiveRabbitListener(queues = "orders.queue")
    public Mono<Void> processOrder(OrderEvent event, 
                                   Channel channel, 
                                   @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        
        return Mono.fromCallable(() -> {
                // Обработка события
                return processOrderInternal(event);
            })
            .doOnSuccess(result -> {
                // Ручное подтверждение
                channel.basicAck(tag, false);
            })
            .doOnError(error -> {
                // Логика повторной обработки
                channel.basicNack(tag, false, shouldRequeue(error));
            })
            .then();
    }
}

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-aot</artifactId>
</dependency>

<dependency>
    <groupId>org.graalvm.buildtools</groupId>
    <artifactId>native-maven-plugin</artifactId>
</dependency>

@NativeHint(
    types = {
        @TypeHint(types = {
            com.rabbitmq.client.impl.AMQConnection.class,
            com.rabbitmq.client.impl.recovery.AutorecoveringConnection.class,
            // Критичные классы для RabbitMQ клиента
        })
    },
    initialization = {
        @InitializationHint(
            types = com.rabbitmq.client.impl.AMQConnection.class,
            initTime = InitializationTime.BUILD
        )
    }
)
@Configuration
public class RabbitMQNativeConfig {
    
    // Явное объявление бинов, необходимых в runtime
    @Bean
    @RuntimeHint
    public ConnectionFactory connectionFactory() {
        // Фабрика должна создаваться с явными типами
        return new CachingConnectionFactory();
    }
}

management:
  endpoints:
    web:
      exposure:
        include: health,metrics,prometheus
      
  endpoint:
    health:
      show-details: always
      probes:
        enabled: true
      
  health:
    rabbit:
      enabled: true
      
    livenessstate:
      enabled: true
    readinessstate:
      enabled: true

@Component
public class RabbitMQBusinessHealthIndicator 
    implements HealthIndicator {
    
    private final RabbitAdmin rabbitAdmin;
    private final MeterRegistry meterRegistry;
    
    public RabbitMQBusinessHealthIndicator(RabbitAdmin rabbitAdmin,
                                          MeterRegistry meterRegistry) {
        this.rabbitAdmin = rabbitAdmin;
        this.meterRegistry = meterRegistry;
    }
    
    @Override
    public Health health() {
        try {
            // Проверка критичных очередей
            Properties queueProperties = rabbitAdmin.getQueueProperties("orders.queue");
            
            if (queueProperties == null) {
                return Health.down()
                    .withDetail("error", "Critical queue not found")
                    .build();
            }
            
            int messageCount = (int) queueProperties.get("QUEUE_MESSAGE_COUNT");
            
            // Проверка на переполнение
            if (messageCount > 10000) {
                return Health.down()
                    .withDetail("warning", "Queue backlog too high")
                    .withDetail("messageCount", messageCount)
                    .build();
            }
            
            return Health.up()
                .withDetail("queueStatus", "healthy")
                .withDetail("messageCount", messageCount)
                .build();
                
        } catch (Exception e) {
            return Health.down(e).build();
        }
    }
}

# Kubernetes deployment конфигурация
apiVersion: apps/v1
kind: Deployment
spec:
  template:
    spec:
      containers:
      - name: order-service
        livenessProbe:
          httpGet:
            path: /actuator/health/liveness
            port: 8080
          initialDelaySeconds: 90  # Учитываем старт Spring Boot 3.x
          periodSeconds: 15
          failureThreshold: 3
          
        readinessProbe:
          httpGet:
            path: /actuator/health/readiness
            port: 8080
          initialDelaySeconds: 30
          periodSeconds: 5
          successThreshold: 2
          failureThreshold: 5
        
        startupProbe:
          httpGet:
            path: /actuator/health/startup
            port: 8080
          failureThreshold: 30  # Долгий старт допустим
          periodSeconds: 10

@Configuration
@EnableWebSecurity
public class RabbitMQSecurityConfig {
    
    @Bean
    public SecurityFilterChain securityFilterChain(HttpSecurity http) throws Exception {
        http
            .authorizeHttpRequests(authz -> authz
                .requestMatchers("/actuator/**").permitAll()
                .anyRequest().authenticated()
            )
            .oauth2ResourceServer(oauth2 -> oauth2
                .jwt(jwt -> jwt
                    .jwtAuthenticationConverter(jwtAuthenticationConverter())
                )
            );
        
        return http.build();
    }
    
    @Bean
    public JwtAuthenticationConverter jwtAuthenticationConverter() {
        JwtGrantedAuthoritiesConverter converter = new JwtGrantedAuthoritiesConverter();
        converter.setAuthorityPrefix("RABBITMQ_");
        converter.setAuthoritiesClaimName("rabbitmq_roles");
        
        JwtAuthenticationConverter jwtConverter = new JwtAuthenticationConverter();
        jwtConverter.setJwtGrantedAuthoritiesConverter(converter);
        
        return jwtConverter;
    }
}

@Component
public class OAuth2RabbitTemplateDecorator {
    
    private final ReactiveClientRegistrationRepository clientRegistrationRepository;
    
    public OAuth2RabbitTemplateDecorator(
            ReactiveClientRegistrationRepository clientRegistrationRepository) {
        this.clientRegistrationRepository = clientRegistrationRepository;
    }
    
    public Mono<RabbitTemplate> decorateWithOAuth2(RabbitTemplate template) {
        return clientRegistrationRepository
            .findByRegistrationId("rabbitmq")
            .flatMap(clientRegistration -> {
                // Получение OAuth2 токена для RabbitMQ
                return retrieveAccessToken(clientRegistration);
            })
            .map(accessToken -> {
                // Декорация сообщений с токеном
                template.addBeforePublishPostProcessors(message -> {
                    message.getMessageProperties()
                        .setHeader("Authorization", "Bearer " + accessToken.getTokenValue());
                    return message;
                });
                return template;
            });
    }
}

spring:
  cloud:
    vault:
      host: ${VAULT_HOST}
      port: 8200
      scheme: https
      authentication: APPROLE
      app-role:
        role-id: ${VAULT_ROLE_ID}
        secret-id: ${VAULT_SECRET_ID}
      kv:
        backend: kv-v2
        default-context: rabbitmq/production

@Configuration
public class VaultRabbitMQConfig {
    
    @Bean
    @VaultPropertySource(
        value = "rabbitmq/production/credentials",
        renewal = Renewal.ROTATE
    )
    public PropertySource<?> vaultPropertySource() {
        return new MapPropertySource("vault-rabbitmq", Map.of());
    }
    
    @Bean
    @RefreshScope
    public ConnectionFactory connectionFactory(
            @Value("${spring.rabbitmq.vault.username}") String username,
            @Value("${spring.rabbitmq.vault.password}") String password) {
        
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setUsername(username);
        factory.setPassword(password);
        
        // Автоматическое обновление при rotation credentials
        factory.setConnectionListeners(Collections.singletonList(connection -> {
            log.info("RabbitMQ connection updated with new credentials");
        }));
        
        return factory;
    }
}

@Configuration
public class MTLSConfig {
    
    @Bean
    @Scheduled(fixedDelay = 3600000) // Каждый час
    public void rotateMTLSCertificates() {
        // Автоматическая ротация сертификатов mTLS
        // Интеграция с cert-manager или аналогичным решением
    }
    
    @Bean
    public SslContextFactory sslContextFactory(VaultTemplate vaultTemplate) {
        return SslContextFactory.builder()
            .keyManager(keyManagerFromVault(vaultTemplate))
            .trustManager(trustManagerFromVault(vaultTemplate))
            .protocol("TLSv1.3")
            .build();
    }
    
    private KeyManager keyManagerFromVault(VaultTemplate vaultTemplate) {
        // Получение клиентского сертификата из Vault
        VaultResponse response = vaultTemplate.read("pki/issue/rabbitmq-client");
        
        byte[] certificate = Base64.getDecoder()
            .decode(response.getData().get("certificate").toString());
        byte[] privateKey = Base64.getDecoder()
            .decode(response.getData().get("private_key").toString());
        
        return loadKeyManager(certificate, privateKey);
    }
}

public class Task050126 {
    public static void main(String[] args) throws InterruptedException {
        Counter c1 = new Counter();
        Counter c2 = new Counter();

        Thread t1 = new Thread(() -> c1.increment());
        Thread t2 = new Thread(() -> c2.increment());

        t1.start();
        t2.start();

        t1.join();
        t2.join();

        System.out.println(Counter.count);
    }

    static class Counter {
        static int count = 0;
        private final Object lock = new Object();

        public void increment() {
            for (int i = 0; i < 100000; i++) {
                synchronized (lock) {
                    count++;
                }
            }
        }
    }
}

import java.util.concurrent.ConcurrentHashMap;

public class Task060126 {
    public static void main(String[] args) {
        ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();

        Integer result = map.computeIfAbsent("a", k -> map.computeIfAbsent("a", k2 -> 2));

        System.out.println(result);
    }
}

@Service
public class TaskProducer {
    private final RabbitTemplate rabbitTemplate;
    
    @Value("${app.queues.task-queue}")
    private String taskQueue;
    
    public void sendTask(Task task) {
        rabbitTemplate.convertAndSend(taskQueue, task, message -> {
            // Установка приоритета задачи
            message.getMessageProperties().setPriority(task.getPriority());
            // Время жизни сообщения
            message.getMessageProperties().setExpiration("3600000"); // 1 час
            return message;
        });
    }
}

@Component
public class TaskWorker {
    
    @RabbitListener(queues = "${app.queues.task-queue}")
    public void processTask(Task task, Channel channel, 
                           @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        try {
            // Обработка задачи
            boolean success = executeTask(task);
            
            if (success) {
                // Подтверждение успешной обработки
                channel.basicAck(deliveryTag, false);
                log.info("Task {} processed successfully", task.getId());
            } else {
                // Отказ без повторной очереди (перемещение в DLQ)
                channel.basicNack(deliveryTag, false, false);
                log.error("Task {} failed, moved to DLQ", task.getId());
            }
        } catch (Exception e) {
            // При ошибке - повторная очередь
            channel.basicNack(deliveryTag, false, true);
            log.error("Error processing task {}, requeued", task.getId(), e);
        }
    }
    
    private boolean executeTask(Task task) {
        // Логика обработки задачи
        return true;
    }
}

@Configuration
public class WorkQueueConfig {
    
    @Bean
    public Queue taskQueue() {
        return QueueBuilder.durable("tasks.queue")
            .withArgument("x-max-priority", 10) // Поддержка приоритетов
            .withArgument("x-dead-letter-exchange", "dlx.exchange")
            .build();
    }
    
    @Bean
    public SimpleRabbitListenerContainerFactory workerFactory(
            ConnectionFactory connectionFactory) {
        
        SimpleRabbitListenerContainerFactory factory = 
            new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        
        // Критичная настройка для равномерного распределения
        factory.setPrefetchCount(1); // По одному сообщению на consumer
        factory.setConcurrentConsumers(3); // Три параллельных worker'а
        factory.setMaxConcurrentConsumers(10); // Автомасштабирование при нагрузке
        
        // Ручные подтверждения для контроля
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        
        return factory;
    }
}

@Configuration
public class PubSubConfig {
    
    @Bean
    public FanoutExchange notificationsExchange() {
        return new FanoutExchange("notifications.fanout", true, false);
    }
    
    @Bean
    public Queue auditQueue() {
        return new Queue("notifications.audit.queue", true);
    }
    
    @Bean
    public Queue cacheQueue() {
        return new Queue("notifications.cache.queue", true);
    }
    
    @Bean
    public Queue analyticsQueue() {
        return new Queue("notifications.analytics.queue", true);
    }
    
    @Bean
    public Binding auditBinding() {
        return BindingBuilder.bind(auditQueue())
            .to(notificationsExchange());
    }
    
    @Bean
    public Binding cacheBinding() {
        return BindingBuilder.bind(cacheQueue())
            .to(notificationsExchange());
    }
    
    @Bean
    public Binding analyticsBinding() {
        return BindingBuilder.bind(analyticsQueue())
            .to(notificationsExchange());
    }
}

@Service
public class NotificationPublisher {
    private final RabbitTemplate rabbitTemplate;
    
    public void publishNotification(Notification notification) {
        // Отправка в fanout exchange - все подписчики получат копию
        rabbitTemplate.convertAndSend(
            "notifications.fanout", 
            "", // routing key игнорируется для fanout
            notification
        );
    }
}

@Component
public class AuditNotificationConsumer {
    
    @RabbitListener(queues = "notifications.audit.queue")
    public void handleNotification(Notification notification) {
        // Каждый consumer обрабатывает свою копию сообщения
        auditService.logEvent(notification);
    }
}

@Configuration
public class TopicRoutingConfig {
    
    @Bean
    public TopicExchange stockExchange() {
        return new TopicExchange("stock.topic", true, false);
    }
    
    @Bean
    public Queue usdStockQueue() {
        return new Queue("stock.usd.queue", true);
    }
    
    @Bean
    public Queue eurStockQueue() {
        return new Queue("stock.eur.queue", true);
    }
    
    @Bean
    public Queue allStockQueue() {
        return new Queue("stock.all.queue", true);
    }
    
    @Bean
    public Binding usdBinding() {
        // Будет получать: stock.usd.nyse, stock.usd.nasdaq
        return BindingBuilder.bind(usdStockQueue())
            .to(stockExchange())
            .with("stock.usd.*");
    }
    
    @Bean
    public Binding eurBinding() {
        // Будет получать: stock.eur.lse, stock.eur.euronext
        return BindingBuilder.bind(eurStockQueue())
            .to(stockExchange())
            .with("stock.eur.*");
    }
    
    @Bean
    public Binding allStocksBinding() {
        // Будет получать все сообщения о stock
        return BindingBuilder.bind(allStockQueue())
            .to(stockExchange())
            .with("stock.#");
    }
}

@Service
public class StockPublisher {
    private final RabbitTemplate rabbitTemplate;
    
    public void publishUSDPrice(String exchange, BigDecimal price) {
        String routingKey = "stock.usd." + exchange.toLowerCase();
        StockUpdate update = new StockUpdate("USD", exchange, price);
        
        rabbitTemplate.convertAndSend(
            "stock.topic", 
            routingKey, 
            update
        );
    }
    
    public void publishEURPrice(String exchange, BigDecimal price) {
        String routingKey = "stock.eur." + exchange.toLowerCase();
        StockUpdate update = new StockUpdate("EUR", exchange, price);
        
        rabbitTemplate.convertAndSend(
            "stock.topic", 
            routingKey, 
            update
        );
    }
}

@Service
public class CalculatorRpcClient {
    private final RabbitTemplate rabbitTemplate;
    
    public CalculatorRpcClient(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
        
        // Настройка reply listener
        this.rabbitTemplate.setReplyTimeout(30000);
        this.rabbitTemplate.setUseDirectReplyToContainer(false);
    }
    
    public BigDecimal calculate(CalculationRequest request) {
        // Создание correlation ID
        String correlationId = UUID.randomUUID().toString();
        
        // Отправка запроса и ожидание ответа
        CalculationResponse response = (CalculationResponse) 
            rabbitTemplate.convertSendAndReceive(
                "rpc.exchange",
                "calculator.rpc",
                request,
                message -> {
                    message.getMessageProperties()
                        .setCorrelationId(correlationId);
                    message.getMessageProperties()
                        .setReplyTo("amq.rabbitmq.reply-to"); // Временная очередь
                    return message;
                }
            );
        
        if (response == null) {
            throw new RuntimeException("RPC timeout");
        }
        
        return response.getResult();
    }
}

@Component
public class CalculatorRpcServer {
    
    @RabbitListener(
        bindings = @QueueBinding(
            value = @Queue(value = "rpc.calculator.queue", durable = "true"),
            exchange = @Exchange(value = "rpc.exchange", type = ExchangeTypes.DIRECT),
            key = "calculator.rpc"
        )
    )
    public CalculationResponse calculate(CalculationRequest request,
                                        Message message) {
        
        String correlationId = message.getMessageProperties()
            .getCorrelationId();
        String replyTo = message.getMessageProperties()
            .getReplyTo();
        
        log.debug("Processing RPC request {} from {}", 
                 correlationId, replyTo);
        
        // Выполнение расчета
        BigDecimal result = performCalculation(request);
        
        // Возврат результата с сохранением correlationId
        CalculationResponse response = new CalculationResponse(result);
        
        // CorrelationId автоматически копируется в ответ
        return response;
    }
    
    private BigDecimal performCalculation(CalculationRequest request) {
        // Логика расчета
        return BigDecimal.ZERO;
    }
}

// Producer без подтверждений
@Bean
public RabbitTemplate atMostOnceTemplate(ConnectionFactory cf) {
    RabbitTemplate template = new RabbitTemplate(cf);
    template.setMandatory(false);
    
    // Отключение publisher confirms
    ((CachingConnectionFactory) cf)
        .setPublisherConfirmType(CachingConnectionFactory.ConfirmType.NONE);
    
    return template;
}

// Consumer с auto-ack
@RabbitListener(queues = "at.most.once.queue")
public void handleAutoAck(Message message) {
    // Сообщение уже удалено из очереди при доставке
    // При падении здесь - сообщение потеряно
    processMessage(message);
}

@Component
public class AtLeastOnceConsumer {
    
    @RabbitListener(queues = "orders.queue")
    public void processOrder(Order order, Channel channel,
                            @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        
        try {
            // 1. Обработка заказа
            processOrderSafely(order);
            
            // 2. Подтверждение после успешной обработки
            channel.basicAck(tag, false);
            
            log.info("Order {} processed and acknowledged", order.getId());
            
        } catch (Exception e) {
            // 3. При ошибке - повторная очередь
            channel.basicNack(tag, false, true);
            log.error("Order processing failed, requeued", e);
            
            // Идемпотентность критична!
            // При повторной обработке должна быть сохранена консистентность
        }
    }
    
    private void processOrderSafely(Order order) {
        // Идемпотентная обработка
        if (orderRepository.existsById(order.getId())) {
            log.warn("Duplicate order {}, skipping", order.getId());
            return;
        }
        
        orderRepository.save(order);
        inventoryService.reserveItems(order);
    }
}

@Service
@Transactional
public class ExactlyOnceProcessor {
    
    private final RabbitTemplate rabbitTemplate;
    private final OrderRepository orderRepository;
    
    public void processWithExactlyOnceSemantics(OrderMessage message) {
        // 1. Проверка идемпотентности
        if (processedMessageRepository.existsById(message.getId())) {
            return; // Уже обработано
        }
        
        // 2. Сохранение в outbox паттерн
        OrderOutbox outbox = new OrderOutbox();
        outbox.setMessageId(message.getId());
        outbox.setOrderData(message.getPayload());
        outbox.setStatus(OutboxStatus.PROCESSING);
        
        orderRepository.save(outbox);
        
        // 3. Бизнес-логика в транзакции
        Order order = createOrder(message);
        orderRepository.save(order);
        
        // 4. Обновление статуса в той же транзакции
        outbox.setStatus(OutboxStatus.PROCESSED);
        orderRepository.save(outbox);
        
        // 5. Отправка подтверждения (после коммита транзакции)
        // RabbitTemplate работает после завершения транзакции
        rabbitTemplate.convertAndSend(
            "order.processed.exchange",
            "order.processed",
            new OrderProcessedEvent(order.getId())
        );
        
        // 6. Идемпотентность на стороне получателя подтверждения
    }
    
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void afterCommit(OrderProcessedEvent event) {
        // Только после успешного коммита
        rabbitTemplate.convertAndSend(
            "order.completed.queue",
            event
        );
    }
}

@Configuration
public class PublisherConfirmConfig {
    
    @Bean
    public CachingConnectionFactory confirmedConnectionFactory() {
        CachingConnectionFactory factory = new CachingConnectionFactory();
        
        // Включение publisher confirms
        factory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        factory.setPublisherReturns(true);
        
        return factory;
    }
    
    @Bean
    public RabbitTemplate confirmedRabbitTemplate(
            CachingConnectionFactory connectionFactory) {
        
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        
        // Callback для подтверждений
        template.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                metrics.increment("publisher.confirms.success");
            } else {
                metrics.increment("publisher.confirms.failure");
                log.error("Message not confirmed: {}", cause);
                
                // Логика повторной отправки
                if (correlationData != null) {
                    retryService.scheduleRetry(correlationData.getId());
                }
            }
        });
        
        // Callback для возвращенных сообщений
        template.setReturnsCallback(returned -> {
            log.error("Message returned: {}", returned.toString());
            returnedMessageService.handleReturned(returned);
        });
        
        return template;
    }
}

public class AcknowledgementExamples {
    
    // 1. Автоматическое подтверждение (ненадежно)
    @RabbitListener(queues = "auto.ack.queue")
    public void autoAck(Message message) {
        // Подтверждение происходит автоматически при возврате метода
    }
    
    // 2. Ручное подтверждение (рекомендуется)
    @RabbitListener(queues = "manual.ack.queue")
    public void manualAck(Order order, Channel channel,
                         @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        
        try {
            process(order);
            channel.basicAck(tag, false); // Положительное подтверждение
        } catch (BusinessException e) {
            // Не requeue для бизнес-ошибок
            channel.basicNack(tag, false, false);
        } catch (TechnicalException e) {
            // Requeue для технических ошибок
            channel.basicNack(tag, false, true);
        }
    }
    
    // 3. Подтверждение с транзакцией
    @Transactional
    @RabbitListener(queues = "transactional.queue")
    public void transactionalAck(Payment payment, Channel channel,
                                @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        
        // Сохранение в БД
        paymentRepository.save(payment);
        
        // Подтверждение произойдет только после коммита транзакции
        // При откате транзакции - сообщение вернется в очередь
    }
}

@Service
public class TransactionalMessageService {
    
    private final RabbitTemplate rabbitTemplate;
    
    @Transactional
    public void processInTransaction(Order order) {
        // 1. Сохранение в базу данных
        orderRepository.save(order);
        
        // 2. Отправка сообщения в той же транзакции
        rabbitTemplate.execute(channel -> {
            // Начало транзакции RabbitMQ
            channel.txSelect();
            
            try {
                // Отправка сообщения
                channel.basicPublish(
                    "orders.exchange",
                    "order.created",
                    null,
                    serialize(order)
                );
                
                // Коммит транзакции RabbitMQ
                channel.txCommit();
                
                return null;
            } catch (Exception e) {
                // Откат транзакции RabbitMQ
                channel.txRollback();
                throw new RuntimeException("Transaction failed", e);
            }
        });
        
        // 3. Если произойдет исключение здесь, 
        // откатятся и БД и сообщение RabbitMQ
        inventoryService.updateStock(order);
    }
}