创建交换机、队列以及绑定关系

1、网页界面创建

2、AmqpAdmin创建

package com.itheima;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.context.annotation.Bean;

@SpringBootTest
class Springboot08SsmpApplicationTests {

    @Autowired
    private AmqpAdmin amqpAdmin;
    @Test
    void contextLoads() {
        Queue fanoutSmsQueue = new Queue("9999", true, false, false);
        amqpAdmin.declareQueue(fanoutSmsQueue);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

3、创建bean，工程自动创建

package com.itheima.config;

import com.itheima.producer.ConfirmCallback;
import com.itheima.producer.ReturnCallback;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;

/**
 * @author xx
 * @date 2023/9/22
 * springboot自动读取yml文件自动配置，这里可删
 * 定义完成后 rabbitmq服务器会自动创建交换机和队列以及绑定关系
 * 在Spring启动时，利用Spring Bean管理工厂BeanFactory接口，实现动态创建交换机、队列、交换机和队列的绑定关系，让我们无需进行重复的编码工作。
 */
@Slf4j
@Configuration
public class RabbitMQConfig {
    /**
     * 队列
     */
    public static final String QUEUE1 = "atguigu";
    public static final String QUEUE2 = "atguigu.news";
    public static final String QUEUE3 = "atguigu.emps";
    public static final String QUEUE4 = "gulixueyuan.news";

    /**
     * 定义交换机名称
     */
    public static final String EXCHANGE_DIRECT_NAME = "exchange.direct";
    public static final String EXCHANGE_FANOUT_NAME = "exchange.fanout";
    public static final String EXCHANGE_TOPIC_NAME = "exchange.topic";
    /**
     * 设置路由key
     * #匹配0个或多个单词，*匹配一个单词
     */
    public static final String ROUTINGKEY1 = "atguigu";
    public static final String ROUTINGKEY2 = "atguigu.news";
    public static final String ROUTINGKEY3 = "atguigu.emps";
    public static final String ROUTINGKEY4 = "gulixueyuan.news";
    public static final String ROUTINGKEY5 = "atguigu.#";
    public static final String ROUTINGKEY6 = "*.news";

    /**
     * 定义 直连交换机
     */
    @Bean("directExchange")
    public DirectExchange directExchange() {
        //参数 交换机名称
        return new DirectExchange(EXCHANGE_DIRECT_NAME, true, false);
    }

    /**
     * 定义 扇形交换机
     */
    @Bean("fanoutExchange")
    public FanoutExchange fanoutExchange() {
        //参数 交换机名称
        return new FanoutExchange(EXCHANGE_FANOUT_NAME, true, false);
    }

    /**
     * 定义 主题交换机
     */
    @Bean("topicExchange")
    public TopicExchange topicExchange() {
        //参数 交换机名称
        return new TopicExchange(EXCHANGE_TOPIC_NAME, true, false);
    }

    /**
     * 创建队列
     * 参数一:队列名称
     * 参数二durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
     * 参数三exclusive:默认也是false，是否独占队列
     * 参数四autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
     */
    @Bean("queue1")
    public Queue queue1() {
        return new Queue(QUEUE1, true, false, false);
    }

    @Bean("queue2")
    public Queue queue2() {
        return new Queue(QUEUE2, true, false, false);
    }

    @Bean("queue3")
    public Queue queue3() {
        return new Queue(QUEUE3, true, false, false);
    }

    @Bean("queue4")
    public Queue queue4() {
        return new Queue(QUEUE4, true, false, false);
    }

    /**
     * 队列绑定交换机
     *
     * @param queue         队列注入到容器的id，也就是方法名 Queue1
     * @param directExchange 交换机注入到容器的id，也就是方法名 directExchange
     * @return
     */
    @Bean
    public Binding bindingQueue1DirectExchange(@Qualifier("queue1") Queue queue, @Qualifier("directExchange") DirectExchange directExchange) {
        return BindingBuilder.bind(queue).to(directExchange).with(ROUTINGKEY1);
    }

    @Bean
    public Binding bindingQueue2DirectExchange(@Qualifier("queue2") Queue queue, @Qualifier("directExchange") DirectExchange directExchange) {
        return BindingBuilder.bind(queue).to(directExchange).with(ROUTINGKEY2);
    }

    @Bean
    public Binding bindingQueue3DirectExchange(@Qualifier("queue3") Queue queue, @Qualifier("directExchange") DirectExchange directExchange) {
        return BindingBuilder.bind(queue).to(directExchange).with(ROUTINGKEY3);
    }

    @Bean
    public Binding bindingQueue4DirectExchange(@Qualifier("queue4") Queue queue, @Qualifier("directExchange") DirectExchange directExchange) {
        return BindingBuilder.bind(queue).to(directExchange).with(ROUTINGKEY4);
    }

    @Bean
    public Binding bindingQueue1FanoutExchange(@Qualifier("queue1") Queue queue, @Qualifier("fanoutExchange") FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(queue).to(fanoutExchange);
    }

    @Bean
    public Binding bindingQueue2FanoutExchange(@Qualifier("queue2") Queue queue, @Qualifier("fanoutExchange") FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(queue).to(fanoutExchange);
    }

    @Bean
    public Binding bindingQueue3FanoutExchange(@Qualifier("queue3") Queue queue, @Qualifier("fanoutExchange") FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(queue).to(fanoutExchange);
    }

    @Bean
    public Binding bindingQueue4FanoutExchange(@Qualifier("queue4") Queue queue, @Qualifier("fanoutExchange") FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(queue).to(fanoutExchange);
    }

    @Bean
    public Binding bindingQueue1TopicExchange(@Qualifier("queue1") Queue queue, @Qualifier("topicExchange") TopicExchange topicExchange) {
        return BindingBuilder.bind(queue).to(topicExchange).with(ROUTINGKEY5);
    }

    @Bean
    public Binding bindingQueue2TopicExchange(@Qualifier("queue2") Queue queue, @Qualifier("topicExchange") TopicExchange topicExchange) {
        return BindingBuilder.bind(queue).to(topicExchange).with(ROUTINGKEY5);
    }

    @Bean
    public Binding bindingQueue3TopicExchange(@Qualifier("queue3") Queue queue, @Qualifier("topicExchange") TopicExchange topicExchange) {
        return BindingBuilder.bind(queue).to(topicExchange).with(ROUTINGKEY5);
    }

    @Bean
    public Binding bindingQueue4TopicExchange(@Qualifier("queue4") Queue queue, @Qualifier("topicExchange") TopicExchange topicExchange) {
        return BindingBuilder.bind(queue).to(topicExchange).with(ROUTINGKEY6);
    }

}
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
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171

4、@RabbitListener

加在方法上：

/**
 * 队列不存在时，需要创建一个队列，并且与exchange绑定
 */
@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "topic.n1", durable = "true", autoDelete = "false"),
        exchange = @Exchange(value = "topic.e", type = ExchangeTypes.TOPIC),
        key = "r"))
public void consumerNoQueue(String data) {
    System.out.println("consumerNoQueue: " + data);
}
1
2
3
4
5
6
7
8
9
10

加在类上需要@RabbitHandler

package com.itheima.service.impl;

import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.baomidou.mybatisplus.extension.service.IService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.itheima.domain.Book;
import com.itheima.mapper.BookMapper;
import com.itheima.service.BookService;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "topic.n1", durable = "true", autoDelete = "true"),
        exchange = @Exchange(value = "topic.e", type = ExchangeTypes.TOPIC),
        key = "r"))
@Service
public class BookServiceImpl extends ServiceImpl<BookMapper, Book> implements BookService {

    @Autowired
    private BookMapper bookMapper;
    @Override
    @RabbitHandler
    public Book queryById(Integer id) {
        Book book = bookMapper.selectById(id);
        return book;
    }
}

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