关于java:RabbitMQ-的七种消息传递形式

6次阅读

共计 11013 个字符，预计需要花费 28 分钟才能阅读完成。

@[toc]
明天这篇文章比较简单，来和小伙伴们分享一下 RabbitMQ 的七种消息传递模式。一起来看看。

大部分状况下，咱们可能都是在 Spring Boot 或者 Spring Cloud 环境下应用 RabbitMQ，因而本文我也次要从这两个方面来和大家分享 RabbitMQ 的用法。

一图胜千言，如下：

这张图中波及到如下一些概念：

生产者（Publisher）：公布音讯到 RabbitMQ 中的交换机（Exchange）上。
交换机（Exchange）：和生产者建设连贯并接管生产者的音讯。
消费者（Consumer）：监听 RabbitMQ 中的 Queue 中的音讯。
队列（Queue）：Exchange 将音讯散发到指定的 Queue，Queue 和消费者进行交互。
路由（Routes）：交换机转发音讯到队列的规定。

大家晓得，RabbitMQ 是 AMQP 营垒里的产品，Spring Boot 为 AMQP 提供了自动化配置依赖 spring-boot-starter-amqp，因而首先创立 Spring Boot 我的项目并增加该依赖，如下：

我的项目创立胜利后，在 application.properties 中配置 RabbitMQ 的根本连贯信息，如下：

spring.rabbitmq.host=localhost
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.port=5672

接下来进行 RabbitMQ 配置，在 RabbitMQ 中，所有的音讯生产者提交的音讯都会交由 Exchange 进行再调配，Exchange 会依据不同的策略将音讯散发到不同的 Queue 中。

RabbitMQ 官网介绍了如下几种音讯散发的模式：

这里给出了七种，其中第七种是音讯确认，音讯确认这块松哥之前发过相干的文章，传送门：

四种策略确保 RabbitMQ 音讯发送可靠性！你用哪种？
RabbitMQ 高可用之如何确保音讯胜利生产

所以这里我次要和大家介绍前六种音讯收发形式。

咦？这个咋没有交换机？这个其实是默认的交换机，咱们须要提供一个生产者一个队列以及一个消费者。音讯流传图如下：

来看看代码实现：

先来看看队列的定义：

@Configuration
public class HelloWorldConfig {

    public static final String HELLO_WORLD_QUEUE_NAME = "hello_world_queue";

    @Bean
    Queue queue1() {return new Queue(HELLO_WORLD_QUEUE_NAME);
    }
}

再来看看音讯消费者的定义：

@Component
public class HelloWorldConsumer {@RabbitListener(queues = HelloWorldConfig.HELLO_WORLD_QUEUE_NAME)
    public void receive(String msg) {System.out.println("msg =" + msg);
    }
}

音讯发送：

@SpringBootTest
class RabbitmqdemoApplicationTests {

    @Autowired
    RabbitTemplate rabbitTemplate;


    @Test
    void contextLoads() {rabbitTemplate.convertAndSend(HelloWorldConfig.HELLO_WORLD_QUEUE_NAME, "hello");
    }

}

这个时候应用的其实是默认的直连交换机（DirectExchange），DirectExchange 的路由策略是将音讯队列绑定到一个 DirectExchange 上，当一条音讯达到 DirectExchange 时会被转发到与该条音讯 routing key 雷同的 Queue 上，例如音讯队列名为“hello-queue”，则 routingkey 为“hello-queue”的音讯会被该音讯队列接管。

这种状况是这样的：

一个生产者，一个默认的交换机（DirectExchange），一个队列，两个消费者，如下图：

一个队列对应了多个消费者，默认状况下，由队列对音讯进行平均分配，音讯会被分到不同的消费者手中。消费者能够配置各自的并发能力，进而进步音讯的生产能力，也能够配置手动 ack，来决定是否要生产某一条音讯。

先来看并发能力的配置，如下：

@Component
public class HelloWorldConsumer {@RabbitListener(queues = HelloWorldConfig.HELLO_WORLD_QUEUE_NAME)
    public void receive(String msg) {System.out.println("receive =" + msg);
    }
    @RabbitListener(queues = HelloWorldConfig.HELLO_WORLD_QUEUE_NAME,concurrency = "10")
    public void receive2(String msg) {System.out.println("receive2 =" + msg+"------->"+Thread.currentThread().getName());
    }
}

能够看到，第二个消费者我配置了 concurrency 为 10，此时，对于第二个消费者，将会同时存在 10 个子线程去生产音讯。

启动我的项目，在 RabbitMQ 后盾也能够看到一共有 11 个消费者。

此时，如果生产者发送 10 条音讯，就会一下都被生产掉。

音讯发送形式如下：

@SpringBootTest
class RabbitmqdemoApplicationTests {

    @Autowired
    RabbitTemplate rabbitTemplate;


    @Test
    void contextLoads() {for (int i = 0; i < 10; i++) {rabbitTemplate.convertAndSend(HelloWorldConfig.HELLO_WORLD_QUEUE_NAME, "hello");
        }
    }

}

音讯生产日志如下：

能够看到，音讯都被第一个消费者生产了。然而小伙伴们须要留神，事件并不总是这样（多试几次就能够看到差别），音讯也有可能被第一个消费者生产（只是因为第二个消费者有十个线程一起开动，所以第二个消费者生产的音讯占比更大）。

当然音讯消费者也能够开启手动 ack，这样能够自行决定是否生产 RabbitMQ 发来的音讯，配置手动 ack 的形式如下：

spring.rabbitmq.listener.simple.acknowledge-mode=manual

生产代码如下：

@Component
public class HelloWorldConsumer {@RabbitListener(queues = HelloWorldConfig.HELLO_WORLD_QUEUE_NAME)
    public void receive(Message message,Channel channel) throws IOException {System.out.println("receive="+message.getPayload());
        channel.basicAck(((Long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG)),true);
    }

    @RabbitListener(queues = HelloWorldConfig.HELLO_WORLD_QUEUE_NAME, concurrency = "10")
    public void receive2(Message message, Channel channel) throws IOException {System.out.println("receive2 =" + message.getPayload() + "------->" + Thread.currentThread().getName());
        channel.basicReject(((Long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG)), true);
    }
}

此时第二个消费者回绝了所有音讯，第一个消费者生产了所有音讯。

这就是 Work queues 这种状况。

再来看公布订阅模式，这种状况是这样：

一个生产者，多个消费者，每一个消费者都有本人的一个队列，生产者没有将音讯间接发送到队列，而是发送到了交换机，每个队列绑定交换机，生产者发送的音讯通过交换机，达到队列，实现一个音讯被多个消费者获取的目标。须要留神的是，如果将音讯发送到一个没有队列绑定的 Exchange 下面，那么该音讯将会失落，这是因为在 RabbitMQ 中 Exchange 不具备存储音讯的能力，只有队列具备存储音讯的能力，如下图：

这种状况下，咱们有四种交换机可供选择，别离是：

Direct
Fanout
Topic
Header

我别离来给大家举一个简略例子看下。

DirectExchange 的路由策略是将音讯队列绑定到一个 DirectExchange 上，当一条音讯达到 DirectExchange 时会被转发到与该条音讯 routing key 雷同的 Queue 上，例如音讯队列名为“hello-queue”，则 routingkey 为“hello-queue”的音讯会被该音讯队列接管。DirectExchange 的配置如下：

@Configuration
public class RabbitDirectConfig {
    public final static String DIRECTNAME = "javaboy-direct";
    @Bean
    Queue queue() {return new Queue("hello-queue");
    }
    @Bean
DirectExchange directExchange() {return new DirectExchange(DIRECTNAME, true, false);
    }
    @Bean
    Binding binding() {return BindingBuilder.bind(queue())
                .to(directExchange()).with("direct");
    }
}

首先提供一个音讯队列 Queue，而后创立一个 DirectExchange 对象，三个参数别离是名字，重启后是否仍然无效以及长期未用时是否删除。
创立一个 Binding 对象将 Exchange 和 Queue 绑定在一起。
DirectExchange 和 Binding 两个 Bean 的配置能够省略掉，即如果应用 DirectExchange，能够只配置一个 Queue 的实例即可。

再来看看消费者：

@Component
public class DirectReceiver {@RabbitListener(queues = "hello-queue")
    public void handler1(String msg) {System.out.println("DirectReceiver:" + msg);
    }
}

通过 @RabbitListener 注解指定一个办法是一个音讯生产办法，办法参数就是所接管到的音讯。而后在单元测试类中注入一个 RabbitTemplate 对象来进行音讯发送，如下：

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitmqApplicationTests {
    @Autowired
    RabbitTemplate rabbitTemplate;
    @Test
    public void directTest() {rabbitTemplate.convertAndSend("hello-queue", "hello direct!");
    }
}

最终执行后果如下：

FanoutExchange 的数据交换策略是把所有达到 FanoutExchange 的音讯转发给所有与它绑定的 Queue 上，在这种策略中，routingkey 将不起任何作用，FanoutExchange 配置形式如下：

@Configuration
public class RabbitFanoutConfig {
    public final static String FANOUTNAME = "sang-fanout";
    @Bean
    FanoutExchange fanoutExchange() {return new FanoutExchange(FANOUTNAME, true, false);
    }
    @Bean
    Queue queueOne() {return new Queue("queue-one");
    }
    @Bean
    Queue queueTwo() {return new Queue("queue-two");
    }
    @Bean
    Binding bindingOne() {return BindingBuilder.bind(queueOne()).to(fanoutExchange());
    }
    @Bean
    Binding bindingTwo() {return BindingBuilder.bind(queueTwo()).to(fanoutExchange());
    }
}

在这里首先创立 FanoutExchange，参数含意与创立 DirectExchange 参数含意统一，而后创立两个 Queue，再将这两个 Queue 都绑定到 FanoutExchange 上。接下来创立两个消费者，如下：

@Component
public class FanoutReceiver {@RabbitListener(queues = "queue-one")
    public void handler1(String message) {System.out.println("FanoutReceiver:handler1:" + message);
    }
    @RabbitListener(queues = "queue-two")
    public void handler2(String message) {System.out.println("FanoutReceiver:handler2:" + message);
    }
}

两个消费者别离生产两个音讯队列中的音讯，而后在单元测试中发送音讯，如下：

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitmqApplicationTests {
    @Autowired
    RabbitTemplate rabbitTemplate;
    @Test
    public void fanoutTest() {
        rabbitTemplate
        .convertAndSend(RabbitFanoutConfig.FANOUTNAME, 
                null, "hello fanout!");
    }
}

留神这里发送音讯时不须要 routingkey，指定 exchange 即可，routingkey 能够间接传一个 null。

最终执行日志如下：

TopicExchange 是比较复杂然而也比拟灵便的一种路由策略，在 TopicExchange 中，Queue 通过 routingkey 绑定到 TopicExchange 上，当音讯达到 TopicExchange 后，TopicExchange 依据音讯的 routingkey 将音讯路由到一个或者多个 Queue 上。TopicExchange 配置如下：

@Configuration
public class RabbitTopicConfig {
    public final static String TOPICNAME = "sang-topic";
    @Bean
    TopicExchange topicExchange() {return new TopicExchange(TOPICNAME, true, false);
    }
    @Bean
    Queue xiaomi() {return new Queue("xiaomi");
    }
    @Bean
    Queue huawei() {return new Queue("huawei");
    }
    @Bean
    Queue phone() {return new Queue("phone");
    }
    @Bean
    Binding xiaomiBinding() {return BindingBuilder.bind(xiaomi()).to(topicExchange())
                .with("xiaomi.#");
    }
    @Bean
    Binding huaweiBinding() {return BindingBuilder.bind(huawei()).to(topicExchange())
                .with("huawei.#");
    }
    @Bean
    Binding phoneBinding() {return BindingBuilder.bind(phone()).to(topicExchange())
                .with("#.phone.#");
    }
}

首先创立 TopicExchange，参数和后面的统一。而后创立三个 Queue，第一个 Queue 用来存储和“xiaomi”无关的音讯，第二个 Queue 用来存储和“huawei”无关的音讯，第三个 Queue 用来存储和“phone”无关的音讯。
将三个 Queue 别离绑定到 TopicExchange 上，第一个 Binding 中的“xiaomi.#”示意音讯的 routingkey 但凡以“xiaomi”结尾的，都将被路由到名称为“xiaomi”的 Queue 上，第二个 Binding 中的“huawei.#”示意音讯的 routingkey 但凡以“huawei”结尾的，都将被路由到名称为“huawei”的 Queue 上，第三个 Binding 中的“#.phone.#”则示意音讯的 routingkey 中但凡蕴含“phone”的，都将被路由到名称为“phone”的 Queue 上。

接下来针对三个 Queue 创立三个消费者，如下：

@Component
public class TopicReceiver {@RabbitListener(queues = "phone")
    public void handler1(String message) {System.out.println("PhoneReceiver:" + message);
    }
    @RabbitListener(queues = "xiaomi")
    public void handler2(String message) {System.out.println("XiaoMiReceiver:"+message);
    }
    @RabbitListener(queues = "huawei")
    public void handler3(String message) {System.out.println("HuaWeiReceiver:"+message);
    }
}

而后在单元测试中进行音讯的发送，如下：

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitmqApplicationTests {
    @Autowired
    RabbitTemplate rabbitTemplate;
    @Test
    public void topicTest() {
        rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME,
"xiaomi.news","小米新闻..");
        rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME,
"huawei.news","华为新闻..");
        rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME,
"xiaomi.phone","小米手机..");
        rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME,
"huawei.phone","华为手机..");
        rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME,
"phone.news","手机新闻..");
    }
}

依据 RabbitTopicConfig 中的配置，第一条音讯将被路由到名称为“xiaomi”的 Queue 上，第二条音讯将被路由到名为“huawei”的 Queue 上，第三条音讯将被路由到名为“xiaomi”以及名为“phone”的 Queue 上，第四条音讯将被路由到名为“huawei”以及名为“phone”的 Queue 上，最初一条音讯则将被路由到名为“phone”的 Queue 上。

HeadersExchange 是一种应用较少的路由策略，HeadersExchange 会依据音讯的 Header 将音讯路由到不同的 Queue 上，这种策略也和 routingkey 无关，配置如下：

@Configuration
public class RabbitHeaderConfig {
    public final static String HEADERNAME = "javaboy-header";
    @Bean
    HeadersExchange headersExchange() {return new HeadersExchange(HEADERNAME, true, false);
    }
    @Bean
    Queue queueName() {return new Queue("name-queue");
    }
    @Bean
    Queue queueAge() {return new Queue("age-queue");
    }
    @Bean
    Binding bindingName() {Map<String, Object> map = new HashMap<>();
        map.put("name", "sang");
        return BindingBuilder.bind(queueName())
                .to(headersExchange()).whereAny(map).match();}
    @Bean
    Binding bindingAge() {return BindingBuilder.bind(queueAge())
                .to(headersExchange()).where("age").exists();}
}

这里的配置大部分和后面介绍的一样，差异次要体现的 Binding 的配置上，第一个 bindingName 办法中，whereAny 示意音讯的 Header 中只有有一个 Header 匹配上 map 中的 key/value，就把该音讯路由到名为“name-queue”的 Queue 上，这里也能够应用 whereAll 办法，示意音讯的所有 Header 都要匹配。whereAny 和 whereAll 实际上对应了一个名为 x-match 的属性。bindingAge 中的配置则示意只有音讯的 Header 中蕴含 age，不论 age 的值是多少，都将音讯路由到名为“age-queue”的 Queue 上。

接下来创立两个音讯消费者：

@Component
public class HeaderReceiver {@RabbitListener(queues = "name-queue")
    public void handler1(byte[] msg) {
        System.out.println("HeaderReceiver:name:"
                + new String(msg, 0, msg.length));
    }
    @RabbitListener(queues = "age-queue")
    public void handler2(byte[] msg) {
        System.out.println("HeaderReceiver:age:"
                + new String(msg, 0, msg.length));
    }
}

留神这里的参数用 byte 数组接管。而后在单元测试中创立音讯的发送办法，这里音讯的发送也和 routingkey 无关，如下：

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitmqApplicationTests {
    @Autowired
    RabbitTemplate rabbitTemplate;
    @Test
    public void headerTest() {
        Message nameMsg = MessageBuilder
                .withBody("hello header! name-queue".getBytes())
                .setHeader("name", "sang").build();
        Message ageMsg = MessageBuilder
                .withBody("hello header! age-queue".getBytes())
                .setHeader("age", "99").build();
        rabbitTemplate.send(RabbitHeaderConfig.HEADERNAME, null, ageMsg);
        rabbitTemplate.send(RabbitHeaderConfig.HEADERNAME, null, nameMsg);
    }
}

这里创立两条音讯，两条音讯具备不同的 header，不同 header 的音讯将被发到不同的 Queue 中去。

最终执行成果如下：

这种状况是这样：

一个生产者，一个交换机，两个队列，两个消费者，生产者在创立 Exchange 后，依据 RoutingKey 去绑定相应的队列，并且在发送音讯时，指定音讯的具体 RoutingKey 即可。

如下图：

这个就是依照 routing key 来路由音讯，我这里就不再举例子了，大家能够参考 3.3.1 小结。

这种状况是这样：

一个生产者，一个交换机，两个队列，两个消费者，生产者创立 Topic 的 Exchange 并且绑定到队列中，这次绑定能够通过 * 和 # 关键字，对指定 RoutingKey 内容，编写时留神格局 xxx.xxx.xxx 去编写。

如下图：

这个我也就不举例啦，后面 3.3.3 大节曾经举过例子了，不再赘述。

RPC 这种音讯收发模式，松哥前两天刚刚写了文章和大家介绍，这里就不多说了，传送门：

SpringBoot+RabbitMQ 实现 RPC 调用

这种发送确认松哥之前有写过相干文章，传送门：

四种策略确保 RabbitMQ 音讯发送可靠性！你用哪种？
RabbitMQ 高可用之如何确保音讯胜利生产

好啦，明天这篇文章次要是和小伙伴们整顿了 RabbitMQ 中音讯收发的七种模式，感兴趣的小伙伴能够试试哦～

公众号【江南一点雨】后盾回复 rabbitmqdemo，获取本文案例地址～

正文完

java

发表至： java

2021-11-16

0

关于java:阿里突遭断网断电双11最惊险一幕刚刚曝光

关于java:呵呵面试官问我知不知道CompletionService

JVM-内存布局

js中将字符串转换成json的三种方式

关于云计算:是云舟呀｜一招教你通过焱融云舟ELK-让日志帮你做决策

关于java:RabbitMQ-的七种消息传递形式

1. RabbitMQ 架构简介

2. 筹备工作

3. 音讯收发

3.1 Hello World

3.2 Work queues

3.3 Publish/Subscrite

3.3.1 Direct

3.3.2 Fanout

3.3.3 Topic

3.3.4 Header

3.4 Routing

3.5 Topics

3.6 RPC

3.7 Publisher Confirms

4. 小结

站内搜索