@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.#");
    }
}