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è)置連接參數(shù)，分別是：主機(jī)名、端口號(hào)、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)建隊(duì)列
        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.關(guān)閉通道和連接
        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 交換機(jī)
     * @param routingKey routing key
     * @param msg 消息
     * @param delay 延遲時(shí)間（如果是延遲交換機(jī)，delay 才有效）
     * @param maxRetries 最大重試機(jī)會(huì)
     * @param <T> 消息的對(duì)象類型
     */
    private <T> void send(String exchange, String routingKey, T msg, long delay, int maxRetries){
        log.info("準(zhǔn)備發(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 從始至終都用的是一個(gè) Consumer 對(duì)象，所以作用的都是同一個(gè)計(jì)數(shù)器
            @Override
            public void accept(CorrelationData.Confirm confirm) {
                Optional.ofNullable(confirm).ifPresent(c -> {
                    if(c.isAck()) {
                        log.info("ACK {} 消息成功到達(dá)，{}", correlationData.getId(), c.getReason());
                    } else {
                        log.warn("NACK {} 消息未能到達(dá)，{}", correlationData.getId(), c.getReason());
                        if(retryCount >= maxRetries) {
                            log.error("次數(shù)到達(dá)上限 {}", 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. 定義消息轉(zhuǎn)換器
        Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter(JsonUtil.OBJECT_MAPPER);
        // 2. 配置自動(dòng)創(chuàng)建消息 id，用于識(shí)別不同消息
        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 則直接丟棄了，并不會(huì)發(fā)送者回執(zhí)
    listener:
      simple:
        prefetch: 1 # 預(yù)取為一個(gè)（消費(fèi)完才能拿下一個(gè)）
        concurrency: 2 # 消費(fèi)者最少 2 個(gè)線程
        max-concurrency: 10 # 消費(fèi)者最多 10 個(gè)線程
        auto-startup: true # 為 false 監(jiān)聽者不會(huì)實(shí)時(shí)創(chuàng)建和監(jiān)聽，為 true 監(jiān)聽的過程中，若 queue 不存在，會(huì)再根據(jù)注解進(jìn)行創(chuàng)建，創(chuàng)建后只監(jiān)聽 queue，declare = "false" 才是不自動(dòng)聲明
        default-requeue-rejected: false # 拒絕后不 requeue（成為死信，若沒有綁定死信交換機(jī)，就真的丟了）
        acknowledge-mode: auto # 消費(fèi)者執(zhí)行成功 ack、異常 nack（manual 為手動(dòng)、none 代表無論如何都是 ack）
        retry: # 這個(gè)屬于 spring amqp 的 retry 機(jī)制
          enabled: false # 不開啟失敗重試
#          initial-interval: 1000
#          multiplier: 2
#          max-attempts: 3
#          stateless: true # true 代表沒有狀態(tài)，若有消費(fèi)者包含事務(wù)，這里改為 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 交換機(jī)
     * @param routingKey routing key
     * @param msg 消息
     * @param delay 延遲時(shí)間（如果是延遲交換機(jī)，delay 才有效）
     * @param maxRetries 最大重試機(jī)會(huì)
     * @param <T> 消息的對(duì)象類型
     */
private <T> void send(String exchange, String routingKey, T msg, long delay, int maxRetries){
    log.info("準(zhǔn)備發(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 從始至終都用的是一個(gè) Consumer 對(duì)象，所以作用的都是同一個(gè)計(jì)數(shù)器
        @Override
        public void accept(CorrelationData.Confirm confirm) {
            Optional.ofNullable(confirm).ifPresent(c -> {
                if(c.isAck()) {
                    log.info("ACK {} 消息成功到達(dá)，{}", correlationData.getId(), c.getReason());
                } else {
                    log.warn("NACK {} 消息未能到達(dá)，{}", correlationData.getId(), c.getReason());
                    if(retryCount >= maxRetries) {
                        log.error("次數(shù)到達(dá)上限 {}", 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 對(duì)應(yīng)的隊(duì)列，那在我看來轉(zhuǎn)發(fā)到零個(gè)是合理的現(xiàn)象，但在這里也認(rèn)為是路由失?。∕Q 認(rèn)為消息一定至少要進(jìn)入一個(gè)隊(duì)列，之后才能被處理，這就是可靠性）（反正就是回執(zhí)了，你愛咋處理是你自己的事情）
    @Override
    public void returnedMessage(ReturnedMessage returnedMessage) {
        // 可能一些版本的 mq 會(huì)因?yàn)槭茄訒r(shí)交換機(jī)，導(dǎo)致發(fā)送者回執(zhí)，只要沒有 NACK 這種情況其實(shí)并不是不可靠（其實(shí)我也不知道有沒有版本會(huì)忽略）
        // 但是其實(shí)不忽略也不錯(cuò)，畢竟者本來就是特殊情況，一般交換機(jī)是不存儲(chǔ)的，但是這個(gè)臨時(shí)存儲(chǔ)消息
        // 但這樣也就代表了，延時(shí)后消息路由失敗是沒法再次處理的（因?yàn)槲覀兘唤o延時(shí)交換機(jī)后就不管了，可靠性有 mq 自己保持）
        MessageProperties messageProperties = returnedMessage.getMessage().getMessageProperties();
        // 這里的 message 并不是原本的 message，是額外的封裝，x-delay 在 publish-returns 里面封裝到 receiveDelay 里了
        Integer delay = messageProperties.getReceivedDelay();
        // 如果不是延時(shí)交換機(jī)，卻設(shè)置了 delay 大于 0，是不會(huì)延時(shí)的，所以是其他原因?qū)е碌模ㄒ苑廊f一把消息記錄到日志里）
        if(Objects.nonNull(delay) && delay.compareTo(0) > 0) {
            log.info("交換機(jī) {}, 路由鍵 {} 消息 {} 延遲 {} s", returnedMessage.getExchange(), returnedMessage.getRoutingKey(), messageProperties, TimeUnit.MILLISECONDS.toSeconds(delay));
            return;
        }
        log.warn("publisher-returns 發(fā)送者回執(zhí)(應(yīng)答碼{}，應(yīng)答內(nèi)容{})(消息 {} 成功到達(dá)交換機(jī) {}，但路由失敗，路由鍵為 {})",
                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);
    // 設(shè)置統(tǒng)一的 publisher-returns（confirm 也可以設(shè)置統(tǒng)一的，但最好還是在發(fā)送時(shí)設(shè)置在 future 里）
    // rabbitTemplate 的 publisher-returns 同一時(shí)間只能存在一個(gè)
    // 因?yàn)?publisher confirm 后，其實(shí) exchange 有沒有轉(zhuǎn)發(fā)成功，publisher 沒必要每次發(fā)送都關(guān)注這個(gè) exchange 的內(nèi)部職責(zé)，更多的是“系統(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，成為死信（默認(rèn)）
//        return new ImmediateRequeueMessageRecoverer(); // nack、requeue
//        return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error"); // ack、發(fā)送給指定的交換機(jī)，confirm 機(jī)制需要設(shè)置到 rabbitTemplate 里
    }
}

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