devbox
花生_TL007
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

RabbitMQ之基础入门

作者：花生_TL007 | 2024-05-05 16:08:12

赞

踩

RabbitMQ之基础入门

在 AMQP 中，Producer 将消息发送到 Exchange ，再由 Exchange 将消息路由到一个或多个 Queue 中（或者丢弃）。Exchange 根据 Routing Key 和 Binding Key 将消息路由到 Queue ，目前提供了 Direct、Topic、Fanout、Headers 四种类型。
RabbitMQ 概念
A. Queue
Queue: Queue（队列）是RabbitMQ的内部对象，用于存储消息,RabbitMQ 中的消息都只能存储在 Queue 中,生产消息并最终投递到Queue中, 消费者可以从Queue中获取消息并消费
在这里插入图片描述
B. Exchange
Exchange: 交换器，实际上这在 RabbitMQ 中是不会生产者将消息投递到队列中，而是，生产者将消息发送到Exchange（交换器），由Exchange将消息路由到一个或多个Queue中（或者丢弃）在这里插入图片描述
C. Routing key
Routing key: 生产者在将消息发送给Exchange的时候，一般会指定一个routing key，来指定这个消息的路由规则，而这个routing key需要与Exchange Type及binding key联合使用才能最终生效。 在Exchange Type与binding key固定的情况下（在正常使用时一般这些内容都是固定配置好的），我们的生产者就可以在发送消息给Exchange时，通过指定routing key来决定消息流向哪里。 RabbitMQ为routing key设定的长度限制为255 bytes。
在这里插入图片描述
D. Binding

Binding: RabbitMQ中通过Binding将Exchange与Queue关联起来，这样RabbitMQ就知道如何正确地将消息路由到指定的Queue了。

E. Binding key

Binding key: 在绑定（Binding）Exchange与Queue的同时，一般会指定一个binding key；消费者将消息发送给Exchange时，一般会指定一个routing key；当binding key与routing key相匹配时，消息将会被路由到对应的Queue中。这个将在Exchange Types章节会列举实际的例子加以说明。 在绑定多个Queue到同一个Exchange的时候，这些Binding允许使用相同的binding key。 binding key 并不是在所有情况下都生效，它依赖于Exchange Type，比如fanout类型的Exchange就会无视binding key，而是将消息路由到所有绑定到该Exchange的Queue。

交换机类型

  1. Direct Exchange
    Direct 类型的 Exchange 路由规则比较简单，它会把消息路由到那些 binding key 与 routing key 完全匹配的 Queue 中
    在这里插入图片描述
    我们以 routingKey=“key1” 发送消息到 Direct Exchange ，则消息会路由到 Queue1
    我们以 routingKey=“key2” 或者 routingKey=“key3” 发送消息到 Direct Exchange ，则消息会路由到 Queue2
    如果我们以其它 routingKey 发送消息，则消息不会路由到这两个 Queue 中。
    2. Topic Exchange
    Direct Exchange路由规则，是完全匹配 binding key 与routing key，但这种严格的匹配方式在很多情况下不能满足实际业务需求。 Topic Exchange 在匹配规则上进行了扩展，它与 Direct 类型的Exchange 相似，也是将消息路由到 binding key 与 routing key 相匹配的 Queue 中，但这里的匹配规则有些不同，它约定：

routing key 为一个句点号 “.” 分隔的字符串。我们将被句点号"."分隔开的每一段独立的字符串称为一个单词，例如 “pay.order.key”
binding key 与 routing key 一样也是句点号 “.” 分隔的字符串。
binding key 中可以存在两种特殊字符 * 与 #，用于做模糊匹配。其中 * 用于匹配一个单词，# 用于匹配多个单词（可以是零个）。
在这里插入图片描述
以上图为例

routingKey=“xx.key1.yy” 的消息会同时路由到 Q1 。
routingKey=“xx.key1.key2” 的消息会同时路由到 Q1 与 Q2 。
routingKey=“xx.yy.key2” 的消息会路由到 Q2 。
routingKey=“key3.xx.key2” 的消息会路由到Q2（只会投递给 Q2 一次，虽然这个 routingKey 与 Q2 的两个 bindingKey 都匹配）。
routingKey=“xx.yy.zz”、routingKey=“key1”、routingKey=“xx.key1.zz.key2” 的消息将会被丢弃，因为它们没有匹配任何 bindingKey 。
3. Fanout Exchange
Fanout Exchange 路由规则非常简单，它会把所有发送到该 Exchange 的消息路由到所有与它绑定的 Queue 中
在这里插入图片描述
生产者发送到 Exchange 的所有消息都会路由到图中的两个 Queue，并最终被消费者消费。
总结来说，指定 Exchange ，会路由到多个绑定的 Queue 中。
4. Headers Exchange

Headers Exchange 不依赖于 routing key 与 binding key 的匹配规则来路由消息，而是根据发送的消息内容中的 headers 属性进行匹配。

在绑定 Queue 与 Exchange 时指定一组 headers 键值对。
当消息发送到 Exchange 时，RabbitMQ 会取到该消息的 headers（也是一个键值对的形式），对比其中的键值对是否完全匹配 Queue 与 Exchange 绑定时指定的键值对；如果完全匹配则消息会路由到该 Queue ，否则不会路由到该 Queue 。
下面进行实战操作
A:Direct模式

confi配置类
@Configuration
public class DirectExchangeConfiguration {
    /**
     * 创建一个 Queue
     *
     * @return Queue
     */
    @Bean
    public Queue queue01() {
        // Queue:名字 | durable: 是否持久化 | exclusive: 是否排它 | autoDelete: 是否自动删除
        return new Queue(
                Message01.QUEUE,
                true,
                false,
                false);
    }

    /**
     * 创建 Direct Exchange
     *
     * @return DirectExchange
     */
    @Bean
    public DirectExchange exchange01() {
        // name: 交换机名字 | durable: 是否持久化 | exclusive: 是否排它
        return new DirectExchange(Message01.EXCHANGE,
                true,
                false);
    }

    /**
     * 创建 Binding
     * Exchange：Message01.EXCHANGE
     * Routing key：Message01.ROUTING_KEY
     * Queue：Message01.QUEUE
     *
     * @return Binding
     */
    @Bean
    public Binding binding01() {
        return BindingBuilder
                .bind(queue01()).to(exchange01())
                .with(Message01.ROUTING_KEY);
    }
}
====================================》direct 类型的消息对象
@Data
public class Message01 implements Serializable {

    public static final String QUEUE = "QUEUE_01";

    public static final String EXCHANGE = "EXCHANGE_01";

    public static final String ROUTING_KEY = "ROUTING_KEY_01";

    private String id;
}
===============================>生产者
@Component
public class Producer01 {
    @Resource
    private RabbitTemplate rabbitTemplate;

    public void syncSend(String id) {
        // 创建 Message01 消息
        Message01 message = new Message01();
        message.setId(id);
        // 同步发送消息
        rabbitTemplate.convertAndSend(Message01.EXCHANGE, Message01.ROUTING_KEY, message);
    }

    public void syncSendDefault(String id) {
        // 创建 Message01 消息
        Message01 message = new Message01();
        message.setId(id);
        // 同步发送消息
        rabbitTemplate.convertAndSend(Message01.QUEUE, message);
    }

    @Async
    public ListenableFuture<Void> asyncSend(String id) {
        try {
            // 发送消息
            this.syncSend(id);
            // 返回成功的 Future
            return AsyncResult.forValue(null);
        } catch (Throwable ex) {
            // 返回异常的 Future
            return AsyncResult.forExecutionException(ex);
        }
    }
==============================================>消费者
@Component
@RabbitListener(queues = Message01.QUEUE)
@Slf4j
public class Consumer01 {
    public void onMessage(Message01 message) {
        log.info("[Consumer01 onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
    }
}
======================================》direct的测试类
 @Resource
    Producer01 producer01;

    @SneakyThrows
    @Test
    void syncSend() {
        String id = UUID.randomUUID().toString();
        producer01.syncSend(id);
        log.info("[test producer01 syncSend][id:{}] 发送成功", id);

        TimeUnit.SECONDS.sleep(2);
    }

    @Test
    void syncSendDefault() throws InterruptedException {
        String id = UUID.randomUUID().toString();
        producer01.syncSendDefault(id);
        log.info("[test producer01 syncSendDefault][id:{}] 发送成功", id);

        TimeUnit.SECONDS.sleep(2);
    }

    @Test
    void asyncSend() throws InterruptedException {
        String id = UUID.randomUUID().toString();

        producer01.asyncSend(id).addCallback(new ListenableFutureCallback<Void>() {

            @Override
            public void onFailure(Throwable e) {
                log.info("[testASyncSend][发送编号：[{}] 发送异常]]", id, e);
            }

            @Override
            public void onSuccess(Void aVoid) {
                log.info("[testASyncSend][发送编号：[{}] 发送成功，发送成功]", id);
            }

        });

        log.info("[test producer01 asyncSend][id:{}] 发送成功", id);

        TimeUnit.SECONDS.sleep(2);
    }


  • 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

B:Fanout 扇出 广播模式

==============》Fanout Exchange 示例的配置类
@Configuration
public class FanoutExchangeConfiguration {
    /**
     * 创建 Queue A
     *
     * @return Queue
     */
    @Bean
    public Queue queue03A() {
        // Queue:名字 | durable: 是否持久化 | exclusive: 是否排它 | autoDelete: 是否自动删除
        return new Queue(Message03.QUEUE_A,
                true,
                false,
                false);
    }

    /**
     * 创建 Queue B
     *
     * @return Queue
     */
    @Bean
    public Queue queue03B() {
        return new Queue(Message03.QUEUE_B,
                true,
                false,
                false);
    }

    /**
     * 创建 Fanout Exchange
     *
     * @return FanoutExchange
     */
    @Bean
    public FanoutExchange demo03Exchange() {
        // name: 交换机名字 | durable: 是否持久化 | exclusive: 是否排它
        return new FanoutExchange(Message03.EXCHANGE,
                true,
                false);
    }

    /**
     * 创建 Binding A
     * Exchange：Message03.EXCHANGE
     * Queue：Message03.QUEUE_A
     *
     * @return Binding
     */
    @Bean
    public Binding demo03BindingA() {
        return BindingBuilder
                .bind(queue03A()).to(demo03Exchange());
    }

    /**
     * 创建 Binding B
     * Exchange：Message03.EXCHANGE
     * Queue：Message03.QUEUE_B
     *
     * @return Binding
     */
    @Bean
    public Binding demo03BindingB() {
        return BindingBuilder
                .bind(queue03B()).to(demo03Exchange());
    }
============================》Fanout Exchange 类型消息
@Data
public class Message03 implements Serializable {
    public static final String QUEUE_A = "QUEUE_03_A";
    public static final String QUEUE_B = "QUEUE_03_B";

    public static final String EXCHANGE = "EXCHANGE_03";

    private String id;
}
==============================》生产者
@Component
public class Producer03 {
    @Resource
    private RabbitTemplate rabbitTemplate;

    /**
     * @param id         消息内容
     * @param routingKey key
     */
    public void syncSend(String id) {
        Message03 message = new Message03();
        message.setId(id);
        rabbitTemplate.convertAndSend(Message03.EXCHANGE, null, message);
    }
==============================》消费者
@Component
@Slf4j
public class Consumer03 {
    @RabbitListener(queues = Message03.QUEUE_A)
    public void onMessage1(Message03 message) {
        log.info("[Consumer03 onMessage1][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
    }

    @RabbitListener(queues = Message03.QUEUE_B)
    public void onMessage2(Message03 message) {
        log.info("[Consumer03 onMessage2][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
    }
}
=====================================》Fanout Exchange 测试类
  @Resource
    Producer03 producer03;

    @Test
    void syncSend() throws InterruptedException {
        String id = UUID.randomUUID().toString();
        producer03.syncSend(id);
        log.info("[test producer03 syncSend][id: {}] 发送成功", id);

        TimeUnit.SECONDS.sleep(2);
    }

  • 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

C:Topic 主题模式

========================》TopicExchange 配置
@Configuration
public class TopicExchangeConfiguration {
    /**
     * 创建 Queue
     *
     * @return Queue
     */
    @Bean
    public Queue queue02() {
        // Queue:名字 | durable: 是否持久化 | exclusive: 是否排它 | autoDelete: 是否自动删除
        return new Queue(Message02.QUEUE,
                true,
                false,
                false);
    }

    /**
     * 创建 Topic Exchange
     *
     * @return TopicExchange
     */
    @Bean
    public TopicExchange exchange02() {
        // name: 交换机名字 | durable: 是否持久化 | exclusive: 是否排它
        return new TopicExchange(Message02.EXCHANGE,
                true,
                false);
    }

    /**
     * 创建 Binding
     * Exchange：Message02.EXCHANGE
     * Routing key：Message02.ROUTING_KEY
     * Queue：Message02.QUEUE
     *
     * @return Binding
     */
    @Bean
    public Binding binding02() {
        return BindingBuilder
                .bind(queue02()).to(exchange02())
                .with(Message02.ROUTING_KEY);
    }
===========================================》topic 类型的实体类
@Data
public class Message02 implements Serializable {
    public static final String QUEUE = "QUEUE_02";

    public static final String EXCHANGE = "EXCHANGE_02";

    public static final String ROUTING_KEY = "#.key2.key3";

    private String id;
}
==============================================》生产者 topic 类型
@Component
public class Producer02 {
    @Resource
    private RabbitTemplate rabbitTemplate;

    /**
     * @param id         消息内容
     * @param routingKey key
     */
    public void syncSend(String id, String routingKey) {
        Message02 message = new Message02();
        message.setId(id);
        rabbitTemplate.convertAndSend(Message02.EXCHANGE, routingKey, message);
    }
}
===================================》 topic 消费者
@Component
@RabbitListener(queues = Message02.QUEUE)
@Slf4j
public class Consumer02 {
    @RabbitHandler
    public void onMessage(Message02 message) {
        log.info("[Consumer02 onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
    }
}
=============================================》topic测试类
 @Resource
    Producer02 producer02;

    @Test
    void syncSendSuccess1() throws InterruptedException {
        String id = UUID.randomUUID().toString();
        // 必须要满足  routingKey 规则才能够成功投递
        String routingKey = "aa.key2.key3";
        producer02.syncSend(id, routingKey);
        log.info("[test producer02 syncSendSuccess1][routingKey: {}][id: {}] 发送成功", routingKey, id);

        TimeUnit.SECONDS.sleep(2);
    }

    @Test
    void syncSendSuccess2() throws InterruptedException {
        String id = UUID.randomUUID().toString();
        String routingKey = "aa.bb.key2.key3";
        producer02.syncSend(id, routingKey);
        log.info("[test producer02 syncSendSuccess2][routingKey: {}][id: {}] 发送成功", routingKey, id);

        TimeUnit.SECONDS.sleep(2);
    }

    @Test
    void syncSendFail() throws InterruptedException {
        String id = UUID.randomUUID().toString();
        String routingKey = "aa.key2.zz";
        producer02.syncSend(id, routingKey);
        log.info("[test producer02 syncSendFail][routingKey:{}][id:{}] 发送成功", routingKey, id);


        TimeUnit.SECONDS.sleep(2);
    }

  • 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

以上的是入门的基础的常见部分代码 若需完整代码 可识别二维码后 给您发代码。
在这里插入图片描述

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/花生_TL007/article/detail/539735
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号