RabbitMq的5種模式及實(shí)例解讀

更新時(shí)間：2023年09月25日 09:46:40 作者：某某灬

這篇文章主要介紹了RabbitMq的5種模式及實(shí)例解讀,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教

1、點(diǎn)對(duì)點(diǎn)，簡(jiǎn)單的生產(chǎn)者消費(fèi)者

在這里插入圖片描述

先創(chuàng)建一個(gè)配置類：

@Configuration
public class RabbitConfig {
    public final static String SIMPLE = "simple";   // 隊(duì)列名稱
    @Bean
    public Queue simpleQueue() {	// 返回一個(gè)bean
        return new Queue(SIMPLE, true, false, true);
    }
}

創(chuàng)建消費(fèi)者和生產(chǎn)者：

@Component
public class SimpleRabbit {
    // 簡(jiǎn)單的點(diǎn)對(duì)點(diǎn)模式
    @Autowired
    RabbitTemplate rabbitTemplate;
    // 生產(chǎn)者
    public void send() throws Exception {
        for (int i = 0; i < 5; i++) {
            rabbitTemplate.convertAndSend(RabbitConfig.SIMPLE, "發(fā)送：" + i);
            Thread.sleep(1000);
        }
    }
	// 消費(fèi)者監(jiān)聽SIMPLE隊(duì)列
    @RabbitListener(queues = RabbitConfig.SIMPLE)
    public void comsumer(String msg) {
        System.out.println("comsumer:" + msg);
    }
 }

然后在測(cè)試類測(cè)試一下：

@RunWith(SpringRunner.class)
@SpringBootTest
public class SimpleTest {
    @Autowired
    SimpleRabbit rabbit;
    @Test
    public void test() throws Exception {
        rabbit.send();
    }
}

結(jié)果為：

在這里插入圖片描述

2、工作模式（work）

工作模式和上邊的模式差不多，只是對(duì)于當(dāng)前的隊(duì)列多了一個(gè)消費(fèi)者：

在這里插入圖片描述

我們只用在 SimpleRabbit類中添加一個(gè)消費(fèi)者就可以完成測(cè)試：

@Component
public class SimpleRabbit {
    // 簡(jiǎn)單的點(diǎn)對(duì)點(diǎn)模式
    @Autowired
    RabbitTemplate rabbitTemplate;
    // 生產(chǎn)者
    public void send() throws Exception {
        for (int i = 0; i < 10; i++) {
            rabbitTemplate.convertAndSend(RabbitConfig.SIMPLE, "發(fā)送：" + i);
            Thread.sleep(1000);
        }
    }
    @RabbitListener(queues = RabbitConfig.SIMPLE)
    public void comsumer1(String o) {
        System.out.println("consumer1:" + o);
    }
    @RabbitListener(queues = RabbitConfig.SIMPLE)
    public void comsumer2(String o) {
        System.out.println("consumer2" + o);
    }
}

直接運(yùn)行test的測(cè)試方法：

在這里插入圖片描述

從打印數(shù)據(jù)來開，是采用輪詢的方式消費(fèi)的。

3、發(fā)布/訂閱模式Publish/Subscribe

在這里插入圖片描述

生產(chǎn)者是把消息發(fā)到交換機(jī)，然后交換機(jī)把數(shù)據(jù)發(fā)送到綁定的隊(duì)列上。

如果沒有隊(duì)列與之綁定，信息將會(huì)丟失，交換機(jī)不能存儲(chǔ)信息。消費(fèi)者監(jiān)聽隊(duì)列，有就消費(fèi)。

為了方便，這里的兩個(gè)隊(duì)列后邊兩個(gè)模式就一起用了。

@Configuration
public class RabbitConfig {
    public final static String PUB_SUB_EXCHANGE = "pub_sub"; // 發(fā)布訂閱的交換機(jī)
    public final static String QUEUE_1 = "queue_1"; //隊(duì)列1
    public final static String QUEUE_2 = "queue_2"; //隊(duì)列2
    @Bean("queue_1")
    public Queue QUEUE_1_Queue() {
        return new Queue(QUEUE_1, true, false, true);
    }
    @Bean("queue_2")
    public Queue QUEUE_2_Queue() {
        return new Queue(QUEUE_2, true, false, true);
    }
    @Bean("pub_sub")	// 交換機(jī)
    public Exchange pub_sub_Exchange(){
        return ExchangeBuilder.fanoutExchange(PUB_SUB_EXCHANGE).build();
    }
	// 使交換機(jī)與隊(duì)列綁定
    @Bean
    public Binding binding_QUEUE_1(@Qualifier("queue_1") Queue queue, @Qualifier(PUB_SUB_EXCHANGE) Exchange exchange) {
    	// with是路由key,這模式默認(rèn)為空就好
        return BindingBuilder.bind(queue).to(exchange).with("").noargs();
    }
    @Bean
    public Binding binding_QUEUE_2(@Qualifier("queue_2") Queue queue, @Qualifier(PUB_SUB_EXCHANGE) Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("").noargs();
    }
}

然后定義生產(chǎn)者和消費(fèi)者：

@Component
public class FanoutRabbit {
    @Autowired
    RabbitTemplate template;
    public void producer() {
        for (int i = 0; i < 5; i++) {
        	// ""是路由key,應(yīng)為沒有所以傳的空
            template.convertAndSend(RabbitConfig.PUB_SUB_EXCHANGE, "", i);
        }
    }
    // 隊(duì)列1的消費(fèi)者
    @RabbitListener(queues = RabbitConfig.QUEUE_1)
    public void queues_1(Integer msg) {
        System.out.println("隊(duì)列1的消費(fèi)者：" + msg);
    }
    // 隊(duì)列1的消費(fèi)者
    @RabbitListener(queues = RabbitConfig.QUEUE_2)
    public void queues_2(Integer msg) {
        System.out.println("隊(duì)列2的消費(fèi)者：" + msg);
    }
}

測(cè)試：

 @Autowired
    FanoutRabbit fanoutRabbit;
    @Test
    public void test1()  {
        fanoutRabbit.producer();
    }

結(jié)果：

在這里插入圖片描述

4、routing路由模式

在這里插入圖片描述

交換機(jī)綁定的路由key和隊(duì)列綁定的一樣時(shí)，才發(fā)送。

@Configuration
public class RabbitConfig {
    public final static String ROUTING_EXCHANGE = "routing1"; // 路由的交換機(jī)
    public final static String ROUTING_1 = "routing_1"; //routing隊(duì)列1
    public final static String ROUTING_2 = "routing_2"; //routing隊(duì)列2
    @Bean("routing_1")
    public Queue OUTING_1_Queue() {
        return new Queue(ROUTING_1, true, false, true);
    }
    @Bean("routing_2")
    public Queue ROUTING_2_Queue() {
        return new Queue(ROUTING_2, true, false, true);
    }
    @Bean("routing1")
    public Exchange routingExchange(){
        return ExchangeBuilder.directExchange(ROUTING_EXCHANGE).build();
    }
    @Bean
    public Binding binding_Routing_QUEUE_1(@Qualifier(ROUTING_1) Queue queue, @Qualifier(ROUTING_EXCHANGE) Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(ROUTING_1).noargs();
    }
    @Bean
    public Binding binding_Routing_QUEUE_2(@Qualifier(ROUTING_2) Queue queue, @Qualifier(ROUTING_EXCHANGE) Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(ROUTING_2).noargs();
    }
}

生產(chǎn)者和消費(fèi)者：

@Component
public class RoutingRabbit {
    @Autowired
    RabbitTemplate template;
    public void producer(){
        for (int i = 0; i < 5; i++) {
            template.convertAndSend(RabbitConfig.ROUTING_EXCHANGE, RabbitConfig.ROUTING_1,i);
        }
    }
    // 隊(duì)列1的消費(fèi)者
    @RabbitListener(queues = RabbitConfig.ROUTING_1)
    public void queues_1(Integer msg) {
        System.out.println("隊(duì)列1的消費(fèi)者：" + msg);
    }
    // 隊(duì)列1的消費(fèi)者
    @RabbitListener(queues = RabbitConfig.ROUTING_2)
    public void queues_2(Integer msg) {
        System.out.println("隊(duì)列2的消費(fèi)者：" + msg);
    }
}

測(cè)試：

在這里插入圖片描述

5、topic主題模式

在這里插入圖片描述

設(shè)置隊(duì)列和交換機(jī)的路由key,當(dāng)生產(chǎn)者發(fā)送消息時(shí)，知道路由key，比較有不有合適的key。有就發(fā)布到隊(duì)列上去。

    public final static String TOPIC_1 = "routing_1"; //topic隊(duì)列1
    public final static String TOPIC_2 = "routing_2"; //topic隊(duì)列2
    @Bean(TOPIC_1)
    public Queue TOPIC_1_Queue() {
        return new Queue(TOPIC_1, true, false, true);
    }
    @Bean(TOPIC_2)
    public Queue TOPIC_2_Queue() {
        return new Queue(TOPIC_2, true, false, true);
    }
    @Bean(TOPIC_EXCHANGE)
    public Exchange topicExchange(){
        return ExchangeBuilder.topicExchange(TOPIC_EXCHANGE).build();
    }
    @Bean
    public Binding binding_Topic_QUEUE_1(@Qualifier(TOPIC_1) Queue queue, @Qualifier(TOPIC_EXCHANGE) Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("aa.bb.cc").noargs();
    }
    @Bean
    public Binding binding_Topic_QUEUE_2(@Qualifier(TOPIC_2) Queue queue, @Qualifier(TOPIC_EXCHANGE) Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("aa.#").noargs();
    }

生產(chǎn)者和消費(fèi)者：

@Component
public class TopicRabbit {
    @Autowired
    RabbitTemplate template;
    public void producer() {
        for (int i = 0; i < 5; i++) {
            template.convertAndSend(RabbitConfig.TOPIC_EXCHANGE, "aa.bb", i);
        }
    }
    // 隊(duì)列1的消費(fèi)者
    @RabbitListener(queues = RabbitConfig.TOPIC_1)
    public void queues_1(Integer msg) {
        System.out.println("隊(duì)列1的消費(fèi)者：" + msg);
    }
    // 隊(duì)列1的消費(fèi)者
    @RabbitListener(queues = RabbitConfig.TOPIC_2)
    public void queues_2(Integer msg) {
        System.out.println("隊(duì)列2的消費(fèi)者：" + msg);
    }
}

測(cè)試結(jié)果：