RabbitMq的5種模式及實(shí)例解讀
先建一個(gè)spring boot項(xiàng)目,然后加入RabbitMQ的依賴就和相關(guān)配置。
在pom.xml中加入:
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>在application.yml中加入配置:
spring:
#rabbitmq配置
rabbitmq:
host: 127.0.0.1
port: 5672
username: guest
password: guest
publisher-confirms: true # 開啟消息確認(rèn),這可不用管
virtual-host: /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é)果:
隊(duì)列2的消費(fèi)者:1
隊(duì)列2的消費(fèi)者:0
隊(duì)列2的消費(fèi)者:3
隊(duì)列2的消費(fèi)者:2
隊(duì)列2的消費(fèi)者:4
*號(hào)代表單個(gè)詞語
#代表多個(gè)詞語
其它的和routing沒什么區(qū)別。
總結(jié)
以上為個(gè)人經(jīng)驗(yàn),希望能給大家一個(gè)參考,也希望大家多多支持腳本之家。
相關(guān)文章
Java中將List列表轉(zhuǎn)換為字符串的三種方法
這篇文章主要介紹了如何在 Java中將List 轉(zhuǎn)換為 String,接下來使用Java 8 Streams Collectors api和String.join()方法將帶有逗號(hào)分隔符或自定義分隔符的集合轉(zhuǎn)換為字符串,需要的朋友可以參考下2025-04-04
教你用Java實(shí)現(xiàn)一個(gè)簡(jiǎn)單的代碼生成器
今天給大家?guī)淼氖顷P(guān)于Java的相關(guān)知識(shí),文章圍繞著如何用Java實(shí)現(xiàn)一個(gè)簡(jiǎn)單的代碼生成器展開,文中有非常詳細(xì)的介紹及代碼示例,需要的朋友可以參考下2021-06-06
Java中將多個(gè)PDF文件合并為一個(gè)PDF的方法步驟
這篇文章主要給大家介紹了關(guān)于Java中將多個(gè)PDF文件合并為一個(gè)PDF的方法步驟, Java PDF合并是指將多個(gè)PDF文件合并成一個(gè)PDF文件的過程,需要的朋友可以參考下2023-09-09
Spring數(shù)據(jù)庫(kù)事務(wù)的實(shí)現(xiàn)機(jī)制講解
這篇文章主要介紹了Spring數(shù)據(jù)庫(kù)事務(wù)的實(shí)現(xiàn)機(jī)制講解,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2021-10-10
實(shí)例解析Java設(shè)計(jì)模式編程中的適配器模式使用
本篇文章主要通過實(shí)例對(duì)適配器模式進(jìn)行了詳解,需要的朋友可以參考下2017-04-04

