docker pull rabbitmq:management

docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:management

spring:
  rabbitmq:
    host: localhost # RabbitMQ服務(wù)器地址
    port: 5672 # 端口號
    username: guest # 用戶名
    password: guest # 密碼
    virtual-host: / # 虛擬主機(jī)，相當(dāng)于一個(gè)獨(dú)立的消息隔離空間，可以在同一RabbitMQ服務(wù)器上創(chuàng)建多個(gè)虛擬主機(jī)，每個(gè)虛擬主機(jī)之間相互隔離，擁有自己的隊(duì)列、交換機(jī)和綁定關(guān)系等。
    # 連接池配置 (Spring Boot 3.x 新特性)，設(shè)置連接超時(shí)時(shí)間為5秒。連接池可以復(fù)用連接，減少連接創(chuàng)建和銷毀的開銷，提高系統(tǒng)性能。
    connection-timeout: 5s 
    # 發(fā)布確認(rèn)，設(shè)置為correlated，表示啟用發(fā)布確認(rèn)機(jī)制，并且使用關(guān)聯(lián)數(shù)據(jù)（CorrelationData）來跟蹤消息的確認(rèn)情況。這樣生產(chǎn)者可以知道消息是否成功發(fā)送到交換機(jī)。
    publisher-confirm-type: correlated 
    publisher-returns: true # 啟用發(fā)布返回，當(dāng)消息發(fā)送到交換機(jī)后，如果無法路由到任何隊(duì)列，會將消息返回給生產(chǎn)者。
    template:
      mandatory: true # 當(dāng)mandatory為true時(shí)，如果消息無法路由到隊(duì)列，會觸發(fā)ReturnCallback，將消息返回給生產(chǎn)者；如果為false，消息會被直接丟棄。
    listener:
      simple:
        acknowledge-mode: manual # 手動(dòng)確認(rèn)，設(shè)置為manual表示消費(fèi)者需要手動(dòng)調(diào)用`channel.basicAck`方法來確認(rèn)消息已被成功處理，防止消息丟失。
        retry:
          enabled: true # 啟用監(jiān)聽重試
          max-attempts: 3 # 最大重試次數(shù)
          initial-interval: 2s # 初始重試間隔時(shí)間為2秒

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitMQConfig {
    // 聲明訂單隊(duì)列，設(shè)置隊(duì)列持久化，并且指定死信交換機(jī)和死信路由鍵，消息10秒過期，隊(duì)列最大長度為1000。
    @Bean
    public Queue orderQueue() {
        return QueueBuilder.durable("order.queue")
               .deadLetterExchange("dlx.exchange")
               .deadLetterRoutingKey("order.dl")
               .ttl(10000)  // 10秒過期
               .maxLength(1000)
               .build();
    }
    // 聲明死信隊(duì)列，用于接收從訂單隊(duì)列中過期或被拒絕的消息。
    @Bean
    public Queue dlQueue() {
        return QueueBuilder.durable("dl.order.queue")
               .build();
    }
    // 聲明訂單交換機(jī)，類型為DirectExchange，并且設(shè)置為持久化。DirectExchange會根據(jù)路由鍵將消息直接路由到對應(yīng)的隊(duì)列。
    @Bean
    public DirectExchange orderExchange() {
        return ExchangeBuilder.directExchange("order.exchange")
               .durable(true)
               .build();
    }
    // 聲明死信交換機(jī)，同樣是DirectExchange類型且持久化。
    @Bean
    public DirectExchange dlxExchange() {
        return ExchangeBuilder.directExchange("dlx.exchange")
               .durable(true)
               .build();
    }
    // 聲明訂單隊(duì)列與訂單交換機(jī)的綁定關(guān)系，指定路由鍵為order.routing.key。
    @Bean
    public Binding orderBinding(Queue orderQueue, DirectExchange orderExchange) {
        return BindingBuilder.bind(orderQueue).to(orderExchange).with("order.routing.key");
    }
    // 聲明死信隊(duì)列與死信交換機(jī)的綁定關(guān)系，路由鍵為order.dl。
    @Bean
    public Binding dlBinding(Queue dlQueue, DirectExchange dlxExchange) {
        return BindingBuilder.bind(dlQueue).to(dlxExchange).with("order.dl");
    }
    // 聲明延遲隊(duì)列，使用RabbitMQ延遲消息插件，設(shè)置隊(duì)列持久化，并添加x-delayed-type參數(shù)，指定為direct類型。
    @Bean
    public Queue delayedQueue() {
        return QueueBuilder.durable("delayed.queue")
               .withArgument("x-delayed-type", "direct")
               .build();
    }
    // 聲明延遲交換機(jī)，類型為CustomExchange，并且設(shè)置為持久化，添加x-delayed-type參數(shù)，指定為direct類型。
    @Bean
    public CustomExchange delayedExchange() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        return new CustomExchange("delayed.exchange", "x-delayed-message", true, false, args);
    }
}

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.UUID;
@Service
public class MessageProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private ObjectMapper objectMapper;
    /**
     * 發(fā)送簡單消息
     */
    public void sendSimpleMessage(String message) {
        rabbitTemplate.convertAndSend("order.exchange", "order.routing.key", message);
    }
    /**
     * 發(fā)送對象消息
     */
    public void sendOrderMessage(Order order) {
        rabbitTemplate.convertAndSend("order.exchange", "order.routing.key", order, message -> {
            // 設(shè)置消息屬性
            message.getMessageProperties().setContentType("application/json");
            return message;
        });
    }
    /**
     * 發(fā)送確認(rèn)消息
     */
    public void sendConfirmMessage(String message) {
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend("order.exchange", "order.routing.key", message, correlationData);
    }
    /**
     * 發(fā)送延遲消息
     */
    public void sendDelayedMessage(String message, int delayMillis) {
        Message msg = MessageBuilder.withBody(message.getBytes()).setHeader("x-delay", delayMillis).build();
        rabbitTemplate.send("delayed.exchange", "delayed.routing.key", msg);
    }
}

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@Service
public class MessageProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @PostConstruct
    public void init() {
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                System.out.println("消息發(fā)送成功: " + correlationData);
            } else {
                System.err.println("消息發(fā)送失敗: " + cause);
            }
        });
    }
}

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class OrderConsumer {
    @RabbitListener(queues = "order.queue")
    public void receiveOrder(String message) {
        System.out.println("接收到訂單消息: " + message);
        // 處理訂單消息的業(yè)務(wù)邏輯
        // 比如調(diào)用訂單處理服務(wù)，更新訂單狀態(tài)等
    }
}

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class OrderConsumer {
    @RabbitListener(queues = "order.queue")
    public void receiveOrder(Message message, Channel channel) {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            String msg = new String(message.getBody());
            System.out.println("接收到訂單消息: " + msg);
            // 處理訂單消息的業(yè)務(wù)邏輯
            // 模擬業(yè)務(wù)處理成功
            boolean success = true; 
            if (success) {
                // 手動(dòng)確認(rèn)消息，參數(shù)1：消息的唯一標(biāo)識，參數(shù)2：是否批量確認(rèn)，這里設(shè)置為false表示只確認(rèn)當(dāng)前這一條消息
                channel.basicAck(deliveryTag, false); 
            } else {
                // 拒絕消息，參數(shù)1：消息的唯一標(biāo)識，參數(shù)2：是否批量拒絕，參數(shù)3：是否重新放回隊(duì)列，這里設(shè)置為false表示不重新放回隊(duì)列，消息會進(jìn)入死信隊(duì)列
                channel.basicNack(deliveryTag, false, false); 
            }
        } catch (Exception e) {
            try {
                // 處理異常時(shí)拒絕消息，進(jìn)入死信隊(duì)列
                channel.basicNack(deliveryTag, false, false); 
            } catch (Exception ex) {
                ex.printStackTrace();
            }
            e.printStackTrace();
        }
    }
}

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class MessageController {
    @Autowired
    private MessageProducer messageProducer;
    @GetMapping("/send")
    public String sendMessage(@RequestParam String message) {
        messageProducer.sendSimpleMessage(message);
        return "消息發(fā)送成功: " + message;
    }
}

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import static org.junit.jupiter.api.Assertions.assertTrue;
@SpringBootTest
public class MessageTest {
    @Autowired
    private MessageProducer messageProducer;
    @Test
    public void testSendMessage() {
        String message = "Test message from unit test";
        messageProducer.sendSimpleMessage(message);
        // 這里可以添加更多的斷言來驗(yàn)證消息發(fā)送的結(jié)果，比如確認(rèn)回調(diào)是否被正確調(diào)用等
        assertTrue(true); 
    }
}