1
docker exec <容器名> rabbitmqctl version

1
# 1. 先查一下容器里真实的插件目录
2
docker exec <容器名> rabbitmq-plugins directories -s
3
# Plugin archives directory: /opt/rabbitmq/plugins  ← 这一行就是目标路径
4

5
# 2. 拷贝进去
6
docker cp rabbitmq_delayed_message_exchange-3.13.0.ez <容器名>:/opt/rabbitmq/plugins/
7

8
# 3. 修正文件权限（docker cp 默认 root:root，容器服务以 rabbitmq 用户运行，读不到会启用失败）
9
docker exec -u root <容器名> chown rabbitmq:rabbitmq \
10
    /opt/rabbitmq/plugins/rabbitmq_delayed_message_exchange-3.13.0.ez
11

12
# 4. 启用插件（热启用，不需要重启容器）
13
docker exec <容器名> rabbitmq-plugins enable rabbitmq_delayed_message_exchange

1
my-rabbitmq/
2
├── docker-compose.yaml
3
├── enabled_plugins
4
└── plugins/
5
    └── rabbitmq_delayed_message_exchange-3.13.0.ez

1
[rabbitmq_management,rabbitmq_delayed_message_exchange].

1
services:
2
  rabbitmq:
3
    image: rabbitmq:3.13-management
4
    container_name: rabbitmq
5
    restart: unless-stopped
6
    ports:
7
      - "5672:5672"       # AMQP
8
      - "15672:15672"     # Management UI
9
    environment:
10
      RABBITMQ_DEFAULT_USER: admin
11
      RABBITMQ_DEFAULT_PASS: ChangeMe_StrongPassword
12
    volumes:
13
      # 持久化消息与元数据
14
      - rabbitmq-data:/var/lib/rabbitmq
15
      # 把本地 plugins/ 目录挂到容器插件路径（具体路径用 rabbitmq-plugins directories -s 确认）
16
      - ./plugins:/opt/rabbitmq/community-plugins
17
      # 预启用插件清单
18
      - ./enabled_plugins:/etc/rabbitmq/enabled_plugins:ro
19
    # 通过 RABBITMQ_PLUGINS_DIR 让 RabbitMQ 从默认目录 + 我们挂的目录都查找插件
20
    # 冒号分隔，官方镜像里插件目录为 /opt/rabbitmq/plugins
21
    command: >
22
      bash -c "
23
        export RABBITMQ_PLUGINS_DIR=/opt/rabbitmq/plugins:/opt/rabbitmq/community-plugins &&
24
        rabbitmq-server
25
      "
26

27
volumes:
28
  rabbitmq-data:

1
docker compose up -d
2
docker compose logs -f rabbitmq

1
docker exec rabbitmq rabbitmq-plugins list -e

1
<dependency>
2
    <groupId>org.springframework.boot</groupId>
3
    <artifactId>spring-boot-starter-amqp</artifactId>
4
</dependency>

1
spring:
2
  rabbitmq:
3
    host: 127.0.0.1
4
    port: 5672
5
    username: admin
6
    password: ChangeMe_StrongPassword

1
@Configuration
2
public class RabbitMQConfig {
3

4
    public static final String DELAYED_EXCHANGE = "delayed.exchange";
5
    public static final String DELAYED_QUEUE    = "delayed.queue";
6
    public static final String DELAYED_ROUTING  = "delayed.key";
7

8
    @Bean
9
    public CustomExchange delayedExchange() {
10
        Map<String, Object> args = new HashMap<>();
11
        // x-delayed-type 指定底层路由行为，direct / topic / fanout 均可
12
        args.put("x-delayed-type", "direct");
13
        return new CustomExchange(
14
                DELAYED_EXCHANGE,
15
                "x-delayed-message",   // 插件提供的 exchange 类型
16
                true,                   // durable
17
                false,                  // autoDelete
18
                args
19
        );
20
    }
21

22
    @Bean
23
    public Queue delayedQueue() {
24
        return new Queue(DELAYED_QUEUE, true);
25
    }
26

27
    @Bean
28
    public Binding delayedBinding() {
29
        return BindingBuilder.bind(delayedQueue())
30
                .to(delayedExchange())
31
                .with(DELAYED_ROUTING)
32
                .noargs();
33
    }
34
}

1
@Service
2
@RequiredArgsConstructor
3
public class DelayedMessageProducer {
4

5
    private final RabbitTemplate rabbitTemplate;
6

7
    /**
8
     * 发送延迟消息
9
     * @param payload     消息内容
10
     * @param delayMillis 延迟毫秒数
11
     */
12
    public void sendDelayed(String payload, long delayMillis) {
13
        rabbitTemplate.convertAndSend(
14
                RabbitMQConfig.DELAYED_EXCHANGE,
15
                RabbitMQConfig.DELAYED_ROUTING,
16
                payload,
17
                message -> {
18
                    // x-delay 单位是毫秒；注意 int 类型、最大约 24.8 天
19
                    message.getMessageProperties().setHeader("x-delay", (int) delayMillis);
20
                    return message;
21
                }
22
        );
23
    }
24
}

1
@Component
2
@Slf4j
3
public class DelayedMessageConsumer {
4

5
    @RabbitListener(queues = RabbitMQConfig.DELAYED_QUEUE)
6
    public void onMessage(String payload) {
7
        log.info("收到延迟消息: {} @ {}", payload, Instant.now());
8
    }
9
}

1
producer.sendDelayed("5 秒后送达", 5_000);
2
producer.sendDelayed("1 分钟后送达", 60_000);