mkdir /root/mq
cd /root/mq
docker rm mq-server -f
docker rmi rabbitmq:3.8-management -f
docker volume rm mq-plugins -f
docker pull rabbitmq:3.8-management
# 插件數(shù)據(jù)卷最好還是直接掛載 volume，而不是掛載我們的目錄
docker run \
--name mq-server \
-e RABBITMQ_DEFAULT_USER=xxx \
-e RABBITMQ_DEFAULT_PASS=xxx \
--hostname mq1 \
-v mq-plugins:/plugins \
-p 15672:15672 \
-p 5672:5672 \
-d rabbitmq:3.8-management

public class PublisherTest {
    @Test
    public void testSendMessage() throws IOException, TimeoutException {
        // 1.建立連接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.設置連接參數(shù)，分別是：主機名、端口號、vhost、用戶名、密碼
        factory.setHost("xx.xx.xx.xx");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("xxx");
        factory.setPassword("xxx");
        // 1.2.建立連接
        Connection connection = factory.newConnection();
        // 2.創(chuàng)建通道Channel
        Channel channel = connection.createChannel();
        // 3.創(chuàng)建隊列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);
        // 4.發(fā)送消息
        String message = "hello, rabbitmq!";
        channel.basicPublish("", queueName, null, message.getBytes());
        System.out.println("發(fā)送消息成功：【" + message + "】");
        // 5.關閉通道和連接
        channel.close();
        connection.close();
    }
}

<!--AMQP依賴，包含RabbitMQ-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

@Repository
@RequiredArgsConstructor
@Slf4j
public class RabbitMQSender {
    private final static ThreadPoolExecutor EXECUTOR = ThreadPoolUtil.getIoTargetThreadPool("Rabbit-MQ-Thread");
    private final RabbitTemplate rabbitTemplate;
    @PostConstruct
    public void init() {
        rabbitTemplate.setTaskExecutor(EXECUTOR);
    }
    private final static Function<Throwable, ? extends CorrelationData.Confirm> ON_FAILURE = ex -> {
        log.error("處理 ack 回執(zhí)失敗, {}", ex.getMessage());
        return null;
    };
    private MessagePostProcessor delayMessagePostProcessor(long delay) {
        return message -> {
            // 小于 0 也是立即執(zhí)行
            // setDelay 才是給 RabbitMQ 看的，setReceivedDelay 是給 publish-returns 看的
            message.getMessageProperties().setDelay((int) Math.max(delay, 0));
            return message;
        };
    };
    private CorrelationData newCorrelationData() {
        return new CorrelationData(UUIDUtil.uuid32());
    }
    /**
     * @param exchange 交換機
     * @param routingKey routing key
     * @param msg 消息
     * @param delay 延遲時間（如果是延遲交換機，delay 才有效）
     * @param maxRetries 最大重試機會
     * @param <T> 消息的對象類型
     */
    private <T> void send(String exchange, String routingKey, T msg, long delay, int maxRetries){
        log.info("準備發(fā)送消息，exchange: {}, routingKey: {}, msg: {}, delay: {}s, maxRetries: {}",
                exchange, routingKey, msg, TimeUnit.MILLISECONDS.toSeconds(delay), maxRetries);
        CorrelationData correlationData = newCorrelationData();
        MessagePostProcessor delayMessagePostProcessor = delayMessagePostProcessor(delay);
        correlationData.getFuture().exceptionallyAsync(ON_FAILURE, EXECUTOR).thenAcceptAsync(new Consumer<>() {
            private int retryCount = 0; // 一次 send 從始至終都用的是一個 Consumer 對象，所以作用的都是同一個計數(shù)器
            @Override
            public void accept(CorrelationData.Confirm confirm) {
                Optional.ofNullable(confirm).ifPresent(c -> {
                    if(c.isAck()) {
                        log.info("ACK {} 消息成功到達，{}", correlationData.getId(), c.getReason());
                    } else {
                        log.warn("NACK {} 消息未能到達，{}", correlationData.getId(), c.getReason());
                        if(retryCount >= maxRetries) {
                            log.error("次數(shù)到達上限 {}", maxRetries);
                            return;
                        }
                        retryCount++;
                        log.warn("開始第 {} 次重試", retryCount);
                        CorrelationData cd = newCorrelationData();
                        cd.getFuture().exceptionallyAsync(ON_FAILURE, EXECUTOR).thenAcceptAsync(this, EXECUTOR);
                        rabbitTemplate.convertAndSend(exchange, routingKey, msg, delayMessagePostProcessor, cd);
                    }
                });
            }
        }, EXECUTOR);
        rabbitTemplate.convertAndSend(exchange, routingKey, msg, delayMessagePostProcessor, correlationData);
    }
    public void sendMessage(String exchange, String routingKey, Object msg) {
        send(exchange, routingKey, msg, 0, 0);
    }
    public void sendDelayMessage(String exchange, String routingKey, Object msg, long delay){
        send(exchange, routingKey, msg, delay, 0);
    }
    public void sendWithConfirm(String exchange, String routingKey, Object msg, int maxReties) {
        send(exchange, routingKey, msg, 0, maxReties);
    }
    public void sendDelayMessageWithConfirm(String exchange, String routingKey, Object msg, long delay, int maxReties) {
        send(exchange, routingKey, msg, delay, maxReties);
    }
}

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "xxx"),
    exchange = @Exchange(name = "xxx", delayed = "true"),
    key = {"xxx"}
))
public void xxx(X x) {
}

@Configuration
@RequiredArgsConstructor
@Slf4j
public class MessageConverterConfig {
    @Bean
    public MessageConverter messageConverter(){
        // 1. 定義消息轉換器
        Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter(JsonUtil.OBJECT_MAPPER);
        // 2. 配置自動創(chuàng)建消息 id，用于識別不同消息
        jackson2JsonMessageConverter.setCreateMessageIds(Boolean.TRUE);
        return jackson2JsonMessageConverter;
    }
}

spring:
  rabbitmq:
    host: ${xxx.mq.host} # rabbitMQ 的 ip 地址
    port: ${xxx.mq.port} # 端口
    username: ${xxx.mq.username}
    password: ${xxx.mq.password}
    virtual-host: ${xxx.mq.virtual-host}
    publisher-confirm-type: correlated
    publisher-returns: true
    template:
      mandatory: true # 若是 false 則直接丟棄了，并不會發(fā)送者回執(zhí)
    listener:
      simple:
        prefetch: 1 # 預取為一個（消費完才能拿下一個）
        concurrency: 2 # 消費者最少 2 個線程
        max-concurrency: 10 # 消費者最多 10 個線程
        auto-startup: true # 為 false 監(jiān)聽者不會實時創(chuàng)建和監(jiān)聽，為 true 監(jiān)聽的過程中，若 queue 不存在，會再根據(jù)注解進行創(chuàng)建，創(chuàng)建后只監(jiān)聽 queue，declare = "false" 才是不自動聲明
        default-requeue-rejected: false # 拒絕后不 requeue（成為死信，若沒有綁定死信交換機，就真的丟了）
        acknowledge-mode: auto # 消費者執(zhí)行成功 ack、異常 nack（manual 為手動、none 代表無論如何都是 ack）
        retry: # 這個屬于 spring amqp 的 retry 機制
          enabled: false # 不開啟失敗重試
#          initial-interval: 1000
#          multiplier: 2
#          max-attempts: 3
#          stateless: true # true 代表沒有狀態(tài)，若有消費者包含事務，這里改為 false

private final static ThreadPoolExecutor EXECUTOR = ThreadPoolUtil.getIoTargetThreadPool("Rabbit-MQ-Thread");
private final RabbitTemplate rabbitTemplate;
@PostConstruct
public void init() {
    rabbitTemplate.setTaskExecutor(EXECUTOR);
}
private final static Function<Throwable, ? extends CorrelationData.Confirm> ON_FAILURE = ex -> {
    log.error("處理 ack 回執(zhí)失敗, {}", ex.getMessage());
    return null;
};
private MessagePostProcessor delayMessagePostProcessor(long delay) {
    return message -> {
        // 小于 0 也是立即執(zhí)行
        // setDelay 才是給 RabbitMQ 看的，setReceivedDelay 是給 publish-returns 看的
        message.getMessageProperties().setDelay((int) Math.max(delay, 0));
        return message;
    };
};
private CorrelationData newCorrelationData() {
    return new CorrelationData(UUIDUtil.uuid32());
}
/**
     * @param exchange 交換機
     * @param routingKey routing key
     * @param msg 消息
     * @param delay 延遲時間（如果是延遲交換機，delay 才有效）
     * @param maxRetries 最大重試機會
     * @param <T> 消息的對象類型
     */
private <T> void send(String exchange, String routingKey, T msg, long delay, int maxRetries){
    log.info("準備發(fā)送消息，exchange: {}, routingKey: {}, msg: {}, delay: {}s, maxRetries: {}",
             exchange, routingKey, msg, TimeUnit.MILLISECONDS.toSeconds(delay), maxRetries);
    CorrelationData correlationData = newCorrelationData();
    MessagePostProcessor delayMessagePostProcessor = delayMessagePostProcessor(delay);
    correlationData.getFuture().exceptionallyAsync(ON_FAILURE, EXECUTOR).thenAcceptAsync(new Consumer<>() {
        private int retryCount = 0; // 一次 send 從始至終都用的是一個 Consumer 對象，所以作用的都是同一個計數(shù)器
        @Override
        public void accept(CorrelationData.Confirm confirm) {
            Optional.ofNullable(confirm).ifPresent(c -> {
                if(c.isAck()) {
                    log.info("ACK {} 消息成功到達，{}", correlationData.getId(), c.getReason());
                } else {
                    log.warn("NACK {} 消息未能到達，{}", correlationData.getId(), c.getReason());
                    if(retryCount >= maxRetries) {
                        log.error("次數(shù)到達上限 {}", maxRetries);
                        return;
                    }
                    retryCount++;
                    log.warn("開始第 {} 次重試", retryCount);
                    CorrelationData cd = newCorrelationData();
                    cd.getFuture().exceptionallyAsync(ON_FAILURE, EXECUTOR).thenAcceptAsync(this, EXECUTOR);
                    rabbitTemplate.convertAndSend(exchange, routingKey, msg, delayMessagePostProcessor, cd);
                }
            });
        }
    }, EXECUTOR);
    rabbitTemplate.convertAndSend(exchange, routingKey, msg, delayMessagePostProcessor, correlationData);
}

@Component
@RequiredArgsConstructor
@Slf4j
public class PublisherReturnsCallBack implements RabbitTemplate.ReturnsCallback {
    // 不存在 routing key 對應的隊列，那在我看來轉發(fā)到零個是合理的現(xiàn)象，但在這里也認為是路由失?。∕Q 認為消息一定至少要進入一個隊列，之后才能被處理，這就是可靠性）（反正就是回執(zhí)了，你愛咋處理是你自己的事情）
    @Override
    public void returnedMessage(ReturnedMessage returnedMessage) {
        // 可能一些版本的 mq 會因為是延時交換機，導致發(fā)送者回執(zhí)，只要沒有 NACK 這種情況其實并不是不可靠（其實我也不知道有沒有版本會忽略）
        // 但是其實不忽略也不錯，畢竟者本來就是特殊情況，一般交換機是不存儲的，但是這個臨時存儲消息
        // 但這樣也就代表了，延時后消息路由失敗是沒法再次處理的（因為我們交給延時交換機后就不管了，可靠性有 mq 自己保持）
        MessageProperties messageProperties = returnedMessage.getMessage().getMessageProperties();
        // 這里的 message 并不是原本的 message，是額外的封裝，x-delay 在 publish-returns 里面封裝到 receiveDelay 里了
        Integer delay = messageProperties.getReceivedDelay();
        // 如果不是延時交換機，卻設置了 delay 大于 0，是不會延時的，所以是其他原因導致的（以防萬一把消息記錄到日志里）
        if(Objects.nonNull(delay) && delay.compareTo(0) > 0) {
            log.info("交換機 {}, 路由鍵 {} 消息 {} 延遲 {} s", returnedMessage.getExchange(), returnedMessage.getRoutingKey(), messageProperties, TimeUnit.MILLISECONDS.toSeconds(delay));
            return;
        }
        log.warn("publisher-returns 發(fā)送者回執(zhí)(應答碼{}，應答內容{})(消息 {} 成功到達交換機 {}，但路由失敗，路由鍵為 {})",
                returnedMessage.getReplyCode(), returnedMessage.getReplyText(), returnedMessage.getMessage(),
                returnedMessage.getExchange(), returnedMessage.getRoutingKey());
    }
}

private final static ThreadPoolExecutor EXECUTOR = ThreadPoolUtil.getIoTargetThreadPool("Rabbit-MQ-Thread");
private final RabbitTemplate rabbitTemplate;
private final PublisherReturnsCallBack publisherReturnsCallBack;
@PostConstruct
public void init() {
    rabbitTemplate.setTaskExecutor(EXECUTOR);
    // 設置統(tǒng)一的 publisher-returns（confirm 也可以設置統(tǒng)一的，但最好還是在發(fā)送時設置在 future 里）
    // rabbitTemplate 的 publisher-returns 同一時間只能存在一個
    // 因為 publisher confirm 后，其實 exchange 有沒有轉發(fā)成功，publisher 沒必要每次發(fā)送都關注這個 exchange 的內部職責，更多的是“系統(tǒng)與 MQ 去約定”
    rabbitTemplate.setReturnsCallback(publisherReturnsCallBack);
}

private final static Function<Throwable, ? extends CorrelationData.Confirm> ON_FAILURE = ex -> {
    log.error("處理 ack 回執(zhí)失敗, {}", ex.getMessage());
    return null;
};

@Configuration
@RequiredArgsConstructor
@Slf4j
public class MessageRecovererConfig {
    @Bean
    public MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate) {
        return new RejectAndDontRequeueRecoverer(); // nack、直接 reject 和不 requeue，成為死信（默認）
//        return new ImmediateRequeueMessageRecoverer(); // nack、requeue
//        return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error"); // ack、發(fā)送給指定的交換機，confirm 機制需要設置到 rabbitTemplate 里
    }
}

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: auto # 消費者執(zhí)行成功 ack、異常 nack（manual 為手動、none 代表無論如何都是 ack）
        retry: # 這個屬于 spring amqp 的 retry 機制
          enabled: false # 不開啟失敗重試
          initial-interval: 1000 # 第一次重試時間間隔
          multiplier: 3 # 每次重試間隔的倍數(shù)
          max-attempts: 4 # 最大接受次數(shù)
          stateless: true # true 代表沒有狀態(tài)，若有消費者包含事務，這里改為 false

message.getMessageProperties().setExpiration("1000");

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "delay.queue", durable = "true"),
        exchange = @Exchange(name = "delay.direct", delayed = "true"),
        key = "delay"
))
public void listenDelayMessage(String msg){
    log.info("接收到delay.queue的延遲消息：{}", msg);
}

private MessagePostProcessor delayMessagePostProcessor(long delay) {
    return message -> {
        // 小于 0 也是立即執(zhí)行
        message.getMessageProperties().setDelay((int) Math.max(delay, 0));
        return message;
    };
};

rabbitmqctl set_policy Lazy "^lazy-queue$" '{"queue-mode":"lazy"}' --apply-to queues  

@RabbitListener(bindings = @QueueBinding(
    exchange = @Exchange(name = "xxx"),
    value = @Queue(name = "xxx", arguments = @Argument(name = "x-queue-mode", value = "lazy")),
    key = "xxx"
))

@RabbitListener(bindings = @QueueBinding(
    exchange = @Exchange(name = "xxx"),
    value = @Queue(name = "xxx", arguments = @Argument(name = "x-queue-type", value = "quorum")),
    key = "xxx"
))