RabbitMQ設定クラス

@Configuration
public class PaymentNotificationConfig implements ApplicationContextAware {

    public static final String PAYMENT_EXCHANGE = "payment_exchange";
    public static final String NOTIFICATION_TYPE = "payment_result";
    public static final String NOTIFICATION_QUEUE = "payment_notification_queue";

    @Bean(PAYMENT_EXCHANGE)
    public FanoutExchange paymentExchange() {
        return new FanoutExchange(PAYMENT_EXCHANGE, true, false);
    }

    @Bean(NOTIFICATION_QUEUE)
    public Queue notificationQueue() {
        return QueueBuilder.durable(NOTIFICATION_QUEUE).build();
    }

    @Bean
    public Binding bindNotificationQueue(
        @Qualifier(NOTIFICATION_QUEUE) Queue queue,
        @Qualifier(PAYMENT_EXCHANGE) FanoutExchange exchange
    ) {
        return BindingBuilder.bind(queue).to(exchange);
    }

    @Override
    public void setApplicationContext(ApplicationContext context) throws BeansException {
        RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class);
        MessageService messageService = context.getBean(MessageService.class);
        
        rabbitTemplate.setReturnCallback((message, code, text, exchange, routingKey) -> {
            log.warn("メッセージ配信失敗: コード{}, 原因{}, エクスチェンジ{}, ルーティングキー{}",
                    code, text, exchange, routingKey);
            PaymentMessage paymentMessage = JSON.parseObject(
                message.toString(), PaymentMessage.class);
            messageService.saveMessage(
                paymentMessage.getMessageType(),
                paymentMessage.getBusinessKey1(),
                paymentMessage.getBusinessKey2(),
                paymentMessage.getBusinessKey3()
            );
        });
    }
}

メッセージ送信処理

@Service
public class PaymentNotificationService {

    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    @Autowired
    private MessageService messageService;

    public void sendPaymentNotification(PaymentMessage message) {
        String jsonMessage = JSON.toJSONString(message);
        Message messageObj = MessageBuilder
            .withBody(jsonMessage.getBytes(StandardCharsets.UTF_8))
            .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
            .build();

        CorrelationData correlationData = new CorrelationData(
            message.getId().toString());

        correlationData.getFuture().addCallback(
            result -> {
                if (result.isAck()) {
                    log.debug("通知送信成功: ID{}", correlationData.getId());
                    messageService.removeMessage(message.getId());
                } else {
                    log.error("通知送信失敗: ID{}, 原因{}", 
                        correlationData.getId(), result.getReason());
                }
            },
            ex -> log.error("送信例外: ID{}, 原因{}", 
                correlationData.getId(), ex.getMessage())
        );

        rabbitTemplate.convertAndSend(
            PaymentNotificationConfig.PAYMENT_EXCHANGE,
            "",
            messageObj,
            correlationData
        );
    }
}

アプリケーション設定

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    publisher-confirm-type: correlated
    publisher-returns: true
    template:
      mandatory: true
    listener:
      simple:
        prefetch: 1
        acknowledge-mode: manual
        retry:
          enabled: true
          initial-interval: 1000ms
          multiplier: 1
          max-attempts: 3
          stateless: true

支払い結果問合せ処理

@Service
public class PaymentQueryService {

    public PaymentRecord queryPaymentResult(String paymentId) {
        AlipayClient client = new DefaultAlipayClient(
            AlipayConfig.URL,
            APP_ID,
            APP_PRIVATE_KEY,
            "json",
            AlipayConfig.CHARSET,
            ALIPAY_PUBLIC_KEY,
            AlipayConfig.SIGNTYPE
        );

        AlipayTradeQueryRequest request = new AlipayTradeQueryRequest();
        JSONObject content = new JSONObject();
        content.put("out_trade_no", paymentId);
        request.setBizContent(content.toString());

        try {
            AlipayTradeQueryResponse response = client.execute(request);
            if (response.isSuccess()) {
                return processPaymentResponse(response, paymentId);
            }
        } catch (AlipayApiException e) {
            log.error("支払い結果問合せエラー: {}", paymentId, e);
        }
        return null;
    }

    private PaymentRecord processPaymentResponse(
        AlipayTradeQueryResponse response, String paymentId) {
        
        Map responseData = JSON.parseObject(
            response.getBody(), Map.class);
        Map tradeData = (Map) responseData.get(
            "alipay_trade_query_response");
        
        PaymentStatus status = new PaymentStatus();
        status.setTradeStatus((String) tradeData.get("trade_status"));
        status.setTradeNo((String) tradeData.get("trade_no"));
        status.setOutTradeNo(paymentId);
        status.setAppId(APP_ID);
        
        return updatePaymentStatus(status);
    }
}

メッセージ消費処理

@Service
public class PaymentMessageConsumer {

    @RabbitListener(queues = PaymentNotificationConfig.NOTIFICATION_QUEUE)
    public void consumeMessage(Message message, Channel channel) {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        String messageBody = new String(message.getBody());
        PaymentMessage paymentMessage = JSON.parseObject(
            messageBody, PaymentMessage.class);

        try {
            processCourseSelection(
                paymentMessage.getBusinessKey1(),
                paymentMessage.getBusinessKey2()
            );
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            handleConsumptionError(channel, deliveryTag, paymentMessage, e);
        }
    }

    private void processCourseSelection(String courseId, String courseType) {
        CourseSelection selection = courseSelectionMapper.selectById(courseId);
        if (selection != null && "60201".equals(courseType)) {
            selection.setStatus("701001");
            courseSelectionMapper.updateById(selection);
            initializeCourseTable(selection);
        }
    }
}