RabbitMQ的三大交换器详解

pom文件都是相同的

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.2.1.RELEASE</version><relativePath/> <!-- lookup parent from repository --></parent><groupId>com.bjsxt</groupId><artifactId>spring-boot-direct-consumer</artifactId><version>0.0.1-SNAPSHOT</version><name>spring-boot-direct-consumer</name><description>spring-boot-direct-consumer</description><properties><java.version>1.8</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

Direct 交换器(发布与订阅完全匹配)

需求

修改 Consumer 的配置文件

server.port=8081
spring.rabbitmq.host=192.168.181.133
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=111111#设置交换器的名称
mq.config.exchange=log.direct#设置info队列名称
mq.config.queue.info=log.info#设置info的路由键
mq.config.queue.info.routing.key=log.info.routing.key#设置error队列名称
mq.config.queue.error=log.error#设置error的路由键
mq.config.queue.error.routing.key=log.error.routing.key

修改 Provider 的配置文件

server.port=8080
spring.rabbitmq.host=192.168.181.133
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=111111#设置交换器的名称
mq.config.exchange=log.direct#设置info的路由键
mq.config.queue.info.routing.key=log.info.routing.key#设置error队列名称
mq.config.queue.error=log.error#设置error的路由键
mq.config.queue.error.routing.key=log.error.routing.key

编写 Consumer

InfoReceiver

package com.bjsxt.receive;import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;/**** 消息的接收者*/
@Component
@RabbitListener(bindings = @QueueBinding(value = @Queue(value = "${mq.config.queue.info}",autoDelete = "true"),exchange = @Exchange(value = "${mq.config.exchange}",type = ExchangeTypes.DIRECT),key="${mq.config.queue.info.routing.key}")
)
public class InfoReceiver {/*** 接收消息的方法* 采用消息队列监听机制* @param msg*/@RabbitHandlerpublic void process(String msg){System.out.println("info--receiver=："+msg);}
}

ErrorReceiver

package com.bjsxt.receive;import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;/**** 消息的接收者*/
@Component
@RabbitListener(bindings = @QueueBinding(value = @Queue(value = "${mq.config.queue.error}",autoDelete = "true"),exchange = @Exchange(value = "${mq.config.exchange}",type = ExchangeTypes.DIRECT),key="${mq.config.queue.error.routing.key}")
)
public class ErrorReceiver {/*** 接收消息的方法* 采用消息队列监听机制* @param msg*/@RabbitHandlerpublic void process(String msg){System.out.println("Error--receiver=："+msg);}
}

编写 Provider

Sender

package com.bjsxt.send;import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;/*** 消息发送者*/
@Component
public class Sender {@Autowiredprivate AmqpTemplate amqpTemplate;@Value("${mq.config.exchange}")private String exchange;@Value("${mq.config.queue.error.routing.key}")private String routingkey;/*** 发送消息的方法* @param msg*/public void sendMsg(String msg){/*向消息队列发送消息*//** 参数一：队列的名称* 参数二：发送的消息* */amqpTemplate.convertAndSend(exchange,routingkey,msg);}
}

com.bjsxt.test.RabbitMQTest

package com.bjsxt.test;import com.bjsxt.SpringBootDirectProviderApplication;
import com.bjsxt.send.Sender;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest(classes = SpringBootDirectProviderApplication.class)
public class RabbitMQTest {@Autowiredprivate Sender sender;@Testpublic void queueTest() throws InterruptedException {while (true){Thread.sleep(1000);sender.sendMsg("你好RabbitMQ");}}
}

Topic 交换器(主题，规则匹配)

需求

Consumer

server.port=8083
spring.rabbitmq.host=192.168.181.133
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=111111#设置交换器的名称
mq.config.exchange=log.topic#设置info队列名称
mq.config.queue.info=log.info#设置info队列名称
mq.config.queue.error=log.error#log 队列名称
mq.config.queue.logs=log.all

Provider

server.port=8084
spring.rabbitmq.host=192.168.181.133
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=111111#设置交换器的名称
mq.config.exchange=log.topic

编写 Provider

UserSender （OrderSender以及ProduceSender三者类似）

package com.bjsxt.send;import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;/*** 消息发送者*/
@Component
public class UserSender {@Autowiredprivate AmqpTemplate amqpTemplate;@Value("${mq.config.exchange}")private String exchange;/*** 发送消息的方法* @param msg*/public void sendMsg(String msg){/*向消息队列发送消息*//** 参数一：队列的名称* 参数二：发送的消息* */amqpTemplate.convertAndSend(exchange,"user.log.bug", "user.log.debug....."+msg);amqpTemplate.convertAndSend(exchange,"user.log.info", "user.log.info....."+msg);amqpTemplate.convertAndSend(exchange,"user.log.warn", "user.log.warn....."+msg);amqpTemplate.convertAndSend(exchange,"user.log.error", "user.log.error....."+msg);}
}

编写 Consumer

InfoReceiver

package com.bjsxt.receive;import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;/**** 消息的接收者*/
@Component
@RabbitListener(bindings = @QueueBinding(value = @Queue(value = "${mq.config.queue.info}",autoDelete = "true"),exchange = @Exchange(value = "${mq.config.exchange}",type = ExchangeTypes.TOPIC),key="*.log.info")
)
public class InfoReceiver {/*** 接收消息的方法* 采用消息队列监听机制* @param msg*/@RabbitHandlerpublic void process(String msg){System.out.println("info--receiver=："+msg);}
}

LogReceiver

package com.bjsxt.receive;import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;/**** 消息的接收者*/
@Component
@RabbitListener(bindings = @QueueBinding(value = @Queue(value = "${mq.config.queue.logs}",autoDelete = "true"),exchange = @Exchange(value = "${mq.config.exchange}",type = ExchangeTypes.TOPIC),key="*.log.*")
)
public class LogReceiver {/*** 接收消息的方法* 采用消息队列监听机制* @param msg*/@RabbitHandlerpublic void process(String msg){System.out.println("log--receiver=："+msg);}
}

Fanout 交换器(广播)

Consumer

spring.application.name=springcloud-mq
spring.rabbitmq.host=192.168.181.133
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=111111
#设置交换器的名称
mq.config.exchange=order.fanout
#短信服务队列名称
mq.config.queue.sms=order.sms
#push 服务队列名称
mq.config.queue.push=order.push

Provider

spring.application.name=springcloud-mq
spring.rabbitmq.host=192.168.181.133
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=111111
#设置交换器的名称
mq.config.exchange=order.fanout

接收方的不停之处在于没有了key其他都是一样的，然后exchange类型换成fanout

发送方

rabbitAmqpTemplate.convertAndSend(this.exchange,"", msg);

RabbitMQ的三大交换器详解相关推荐

数据库三大范式详解，部分依赖、完全依赖、传递依赖
数据库三大范式详解以及部分.完全.传递依赖一.第一范式二.第二范式三.第三范式四.部分依赖.完全依赖.传递依赖完结撒花一.第一范式数据库每一列都是不可分的基本数据项(原子数据项) 就比如 ...
IBM虚拟化与云计算研究员吴玉会——虚拟化三大优势详解：共享、优化、管理...
IBM虚拟化与云计算专家- IT热书<虚拟化与云计算>作者组做客51CTO巅峰访谈-2 IBM虚拟化与云计算研究员吴玉会--虚拟化三大优势详解:共享.优化.管理 <虚拟化与云计算&g ...
数据库三大范式详解实例图文教程
title: 数据库三大范式详解实例图文教程 date: 2021-09-04 20:29:31 tags: 数据库 MySQL categories: 数据库 cover: https://cove ...
Python精讲：在Python中遍历字典的三大方法详解
欢迎你来到站长在线的站长学堂学习Python知识,本文学习的是<在Python中遍历字典的三大方法详解>.本知识点主要内容有:使用字典对象的items()方法可以遍历字典的项和字典的&qu ...
RabbitMq的六种模式分析详解
AMQP 即Advanced Message Queuing Protocol,高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计.消息中间件主要用于组件之间的解耦,消息的发送者无需 ...
RabbitMQ 相关概念和方法详解
名词解释 ConnectionFactory: 与 RabbitMQ 服务器连接的管理器. Connection: 与 RabbitMQ 服务器的连接. Channel: 与 Exchange 的连接 ...
直播回顾 | 数据驱动「产品迭代」的三大场景详解
近日,神策数据进行了一场题为<数据驱动产品迭代的三大场景>的专题直播,直播中结合各行业的数据驱动企业的优质实践经验,针对三大产品迭代场景进行了逐一详解.如下为直播的主要内容: 场景一. 产 ...
谷粒商城RabbitMQ锁库存逻辑详解--新理解(长文警告)
前言不废话,上来就说,代码我会放挺多,写过这个项目的自然能懂,如果真的像理解的请认真看哦分析 /*出现的问题:扣减库存成功了,但是由于网络原因超时,出现异常,导致订单事务回滚,库存事务不回滚(解决 ...
JAVA面向对象三大特征详解：(封装，继承，多态)
面向对象语言: 大家经常会听到说JAVA这门语言是面向对象的,但面向对象又是什么呢? 提到面向对象就要引入一个叫做面向过程的概念: 举个例子:把大象装进冰箱,需要几步. 面向过程的做法: 人走到冰箱前 ...

RabbitMQ的三大交换器详解

RabbitMQ的三大交换器详解相关推荐

最新文章

热门文章