1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

import org.springframework.amqp.core.Queue;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

@Configuration

public class StandardQueueConfig {

    public static final String STANDARD_QUEUE = "standard_queue";

    @Bean

    public Queue standardQueue() {

        // 参数说明：

        // 1. 队列名称

        // 2. durable: 是否持久化 (true: 重启后队列依然存在)

        return new Queue(STANDARD_QUEUE, true);

    }

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

import org.springframework.amqp.core.Queue;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

import java.util.HashMap;

import java.util.Map;

@Configuration

public class PriorityQueueConfig {

    public static final String PRIORITY_QUEUE = "priority_queue";

    @Bean

    public Queue priorityQueue() {

        Map<String, Object> args = new HashMap<>();

        // 开启优先级队列，并设置最大优先级为 10（官方建议值为 1-10，最高可设置 255）

        args.put("x-max-priority", 10);

        // 参数说明：名称, durable, exclusive(独占), autoDelete(自动删除), arguments(特殊属性)

        return new Queue(PRIORITY_QUEUE, true, false, false, args);

    }

}

1

2

3

4

rabbitTemplate.convertAndSend("exchange_name", "routing_key", "紧急消息", message -> {

    message.getMessageProperties().setPriority(9); // 设置高优先级

    return message;

});

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

import org.springframework.amqp.core.*;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

import java.util.HashMap;

import java.util.Map;

@Configuration

public class DeadLetterQueueConfig {

    // ==================== 1. 定义死信交换机和死信队列 ====================

    public static final String DLX_EXCHANGE = "dlx_exchange";

    public static final String DEAD_LETTER_QUEUE = "dead_letter_queue";

    public static final String DLX_ROUTING_KEY = "dlx_routing_key";

    @Bean

    public DirectExchange dlxExchange() {

        return new DirectExchange(DLX_EXCHANGE);

    }

    @Bean

    public Queue deadLetterQueue() {

        return new Queue(DEAD_LETTER_QUEUE, true);

    }

    @Bean

    public Binding dlxBinding() {

        return BindingBuilder.bind(deadLetterQueue()).to(dlxExchange()).with(DLX_ROUTING_KEY);

    }

    // ==================== 2. 定义普通业务队列 (并绑定死信机制) ====================

    public static final String BUSINESS_QUEUE = "business_queue";

    @Bean

    public Queue businessQueue() {

        Map<String, Object> args = new HashMap<>();

        // 核心配置：当这个业务队列出现死信时，交给哪个死信交换机处理

        args.put("x-dead-letter-exchange", DLX_EXCHANGE);

        // 核心配置：发给死信交换机时，带上什么路由键

        args.put("x-dead-letter-routing-key", DLX_ROUTING_KEY);

        // 为了演示触发死信，我们给这个业务队列加一个最大长度限制

        // 如果队列堆积超过 5 条，第 6 条就会变成死信，进入 dead_letter_queue

        args.put("x-max-length", 5);

        return new Queue(BUSINESS_QUEUE, true, false, false, args);

    }

}

x-dead-letter-exchange

x-dead-letter-routing-key