概述
概述
RabbitMQ 即一个消息队列，主要是用来实现应用程序的异步和解耦，同时也能起到消息缓冲，消息分发的作用。消息中间件最主要的作用是解耦，中间件最标准的用法是生产者生产消息传送到队列，消费者从队列中拿取消息并处理，生产者不用关心是谁来消费，消费者不用关心谁在生产消息，从而达到解耦的目的。在分布式的系统中，消息队列也会被用在很多其它的方面，比如：分布式事务的支持，RPC的调用等等。RabbitMQ是流行的开源消息队列系统，用erlang语言开发。RabbitMQ是AMQP（高级消息队列协议）的标准实现。如果不熟悉AMQP，直接看RabbitMQ的文档会比较困难。不过RabbitMQ也就几个关键概念，关键词理解更有助于理解其工作方式和原理，RabbitMQ的结构图如下：
关键词
Broker：简单来说就是消息队列服务器实体；
Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列；
Queue：消息队列载体，每个消息都会被投入到一个或多个队列；
Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来；
Routing Key：路由关键字，exchange根据这个关键字进行消息投递；
vhost：虚拟主机，一个broker里可以开设多个vhost，用作不同用户的权限分离；
producer：消息生产者，就是投递消息的程序；
consumer：消息消费者，就是接受消息的程序；
channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务；
消息队列的使用过程步骤归纳如下：
- client连接到消息队列server，打开一个channel;
- client声明一个exchange，并设置相关属性;
- client声明一个queue，并设置相关属性;
- client使用routing key，在exchange和queue之间建立好绑定关系;
- client投递消息到exchange;
exchange接收到消息后，就根据消息的key和配置的binding，进行消息路由，将消息投递到一个或多个队列里。exchange也有几个类型：
①Direct：仅依据 key进行消息投递的叫做Direct交换机，例如，绑定时设置了routing key为”test”，那么客户端提交的消息，只有设置了key为”test”的才会投递到队列；
②Topic：依据key进行模式匹配后进行投递的叫做Topic交换机，符号”#”匹配一个或多个词，符号”*”匹配正好一个词。例如”abc.#”匹配”abc.def.ghi”，”abc.*”只匹配”abc.def”；
③Fanout：存在一种模式下不需要key叫做Fanout交换机，它采取广播模式，一个消息进来时，投递到与该交换机绑定的所有队列，就类似微信公众号的推送，你只要订阅了某个公众号，所有关注这个公众号的微信都会收到推送消息；
④Headers：是根据一个规则进行匹配,在消息队列和交换机绑定的时候会指定一组键值对规则,而发送消息的时候也会指定一组键值对规则,当两组键值对规则相匹配的时候,消息会被发送到匹配的消息队列中。
消息持久化
RabbitMQ支持消息的持久化，也就是数据写在磁盘上，为了数据安全考虑，我想大多数用户都会选择持久化。消息队列持久化包括3个部分：
（1）exchange持久化，在声明时指定durable => 1
（2）queue持久化，在声明时指定durable => 1
（3）消息持久化，在投递时指定delivery_mode => 2（1是非持久化）
如果exchange和queue都是持久化的，那么它们之间的binding也是持久化的。如果exchange和queue两者之间有一个持久化，一个非持久化，就不允许建立绑定。
RabbitMQ使用
基本设置
// 1. 引入pom
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
// 2.配置文件
#配置rabbitmq的安装地址、端口以及账户信息
#注意port 15672是管理端的端口
spring.application.name=spirng-boot-rabbitmq-sender
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
Direct模式简单测试
producer 和 consumer的queue name必须一致，否则会收不到消息。Direct模式相当于一对一模式，producer 发送一个消息被后，会被转发到绑定的消息队列中，然后被一个 consumer接收。
// 队列配置
@Configuration
public class RabbitConfig {
@Bean
public Queue helloQueue() {
return new Queue("hello");
}
}
// producer
public void send(User user) {
System.out.println("Sender object: " + user.toString());
this.rabbitTemplate.convertAndSend("object", user);
}
// consumer
@RabbitHandler
public void process(User user) {
System.out.println("consumer object : " + user);
System.out.println("username:"+user.getUsername());
System.out.println("password:"+user.getPassword());
}
// TEST
@org.junit.Test
public void sendObj() {
User user = new User();
user.setId(1);
user.setUsername("admin");
user.setPassword("1234");
objectSender.send(user);
}
// TEST RESULT
Sender object: com.example.bean.User@fa5f81c
2019-01-06 11:13:40.175 INFO 26456 --- [ Thread-2]
o.s.a.r.l.SimpleMessageListenerContainer : Waiting for workers to finish.
consumer object : com.example.bean.User@7b0f425
username:admin
password:1234
Topic模式
Topic 模式是 RabbitMQ中最灵活的一种方式，可以根据routing_key自由的绑定不同的队列，这就需要单独配置队列对应关系。
1.Topic模式配置类
@Configuration
public class TopicRabbitConfig {
// 一般在配置文件
final static String one= "TOPIC_MESSAGE_ONE";
final static String two= "TOPIC_MESSAGE_TWO";
// 创建队列
@Bean
public Queue queueMessage() {
return new Queue(TopicRabbitConfig.one);
}
// 创建队列
@Bean
public Queue queueMessages() {
return new Queue(TopicRabbitConfig.two);
}
// 将对列绑定到Topic交换器
@Bean
TopicExchange exchange() {
return new TopicExchange("topicExchange");
}
// 将对列绑定到Topic交换器
@Bean
Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {
return BindingBuilder.bind(queueMessage).to(exchange).with("topic.one");
}
// 将对列绑定到Topic交换器 采用#的方式
@Bean
Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {
return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");
}
}
2. 测试类
// producer one
public void send1() {
String context = "hello test one";
System.out.println("Sender : " + context);
this.rabbitTemplate.convertAndSend("topicExchange", "topic.one", context);
}
// producer two
public void send2() {
String context = "hello test two";
System.out.println("Sender : " + context);
this.rabbitTemplate.convertAndSend("topicExchange", "topic.two", context);
}
// consumer
@Component
@RabbitListener(queues = "topic.one")
public class TopicReceiver {
@RabbitHandler
public void process(String message) {
System.out.println("Topic consumer one: " + message);
}
}
/*
* producer 发送send1会匹配到topic.#和topic.one 两个consumer都可以收到消息，
* producer 发送send2只有topic.#可以匹配所有只有consumer two监听到消息
*/
// test
@RunWith(SpringRunner.class)
@SpringBootTest
public class Test {
@Autowired
private TopicSender topicSender;
@org.junit.Test
public void send1() {
topicSender.send1();
}
@org.junit.Test
public void send2() {
topicSender.send2();
}
}Headers 模式
1.headers模式队列配置类
@Configuration
public class HeadersConfig {
@Bean
public Queue creditBankQueue() {
return new Queue("credit.bank");
}
@Bean
public Queue creditFinanceQueue() {
return new Queue("credit.finance");
}
@Bean
public HeadersExchange creditBankExchange() {
return new HeadersExchange("creditBankExchange");
}
@Bean
public HeadersExchange creditFinanceExchange() {
return new HeadersExchange("creditFinanceExchange");
}
// 配置一个routingKey为credit.bank的消息队列并绑定在creditBankExchange交换机上
@Bean
public Binding bindingCreditAExchange(Queue creditBankQueue, HeadersExchange creditBankExchange) {
Map<String,Object> headerValues = new HashMap<>();
headerValues.put("type", "cash");
headerValues.put("aging", "fast");
return BindingBuilder.bind(creditBankQueue).to(creditBankExchange).whereAll(headerValues).match();
}
// 配置一个routingKey为credit.finance的消息队列并绑定在creditFinanceExchange交换机上
@Bean
public Binding bindingCreditBExchange(Queue creditFinanceQueue, HeadersExchange creditFinanceExchange) {
Map<String,Object> headerValues = new HashMap<>();
headerValues.put("type", "cash");
headerValues.put("aging", "fast");
return BindingBuilder.bind(creditFinanceQueue).to(creditFinanceExchange).whereAny(headerValues).match();
}
}2.测试类
// producer
@Component
public class ApiCreditSender {
@Autowired
private AmqpTemplate rabbitTemplate;
public void creditBank(Map<String, Object> head, String msg){
System.out.println("credit.bank send message: "+msg);
rabbitTemplate.convertAndSend("creditBankExchange", "credit.bank", getMessage(head, msg));
}
public void creditFinance(Map<String, Object> head, String msg){
System.out.println("credit.finance send message: "+msg);
rabbitTemplate.convertAndSend("creditFinanceExchange", "credit.finance", getMessage(head, msg));
}
private Message getMessage(Map<String, Object> head, Object msg){
MessageProperties messageProperties = new MessageProperties();
for (Map.Entry<String, Object> entry : head.entrySet()) {
messageProperties.setHeader(entry.getKey(), entry.getValue());
}
MessageConverter messageConverter = new SimpleMessageConverter();
return messageConverter.toMessage(msg, messageProperties);
}
}
// consumer
@Component
public class ApiCreditReceive {
@RabbitHandler
@RabbitListener(queues = "credit.bank")
public void creditBank(String msg) {
System.out.println("credit.bank receive message: "+msg);
}
@RabbitHandler
@RabbitListener(queues = "credit.finance")
public void creditFinance(String msg) {
System.out.println("credit.bank receive message: "+msg);
}
}
// test
@RunWith(SpringRunner.class)
@SpringBootTest
public class Test {
@Autowired
private ApiCreditSender sender;
@org.junit.Test
public void test_creditBank_type() {
Map<String,Object> head = new HashMap<>();
head.put("type", "cash");
sender.creditBank(head, "银行授信(部分匹配)");
}
@org.junit.Test
public void test_creditBank_all() {
Map<String,Object> head = new HashMap<>();
head.put("type", "cash");
head.put("aging", "fast");
sender.creditBank(head, "银行授信(全部匹配)");
}
@org.junit.Test
public void test_creditFinance_type() {
Map<String,Object> head = new HashMap<>();
head.put("type", "cash");
sender.creditFinance(head, "金融公司授信(部分匹配)");
}
@org.junit.Test
public void test_creditFinance_all() {
Map<String,Object> head = new HashMap<>();
head.put("type", "cash");
head.put("aging", "fast");
sender.creditFinance(head, "金融公司授信(全部匹配)");
}
}
// result
// 发送者日志
credit.bank send message: 银行授信(部分匹配)
credit.finance send message: 金融公司授信(全部匹配)
credit.bank send message: 银行授信(全部匹配)
credit.finance send message: 金融公司授信(部分匹配)
// 接收者日志
credit.bank receive message: 银行授信(全部匹配)
credit.bank receive message: 金融公司授信(全部匹配)
credit.bank receive message: 金融公司授信(部分匹配)根据日志中可以看出4个测试方法均已成功发送消息。但是credit.bank监听的队列有一条消息没有接收到。ApiCreditSenderTests.test_creditBank_type()为什么发送的消息，没有被处理？因为在HeadersConfig配置类中，creditBankExchange交换机的匹配规则是完全匹配，即header attribute参数必须完成一致。
Fanout模式
1.Fanout模式队列配置类：Fanout Exchange交换机是转发消息到所有绑定队列（和routingKey没有关系）,即我们熟悉的广播模式或者订阅模式，给Fanout交换机发送消息，绑定了这个交换机的所有队列都收到这个消息。
@Configuration
public class FanoutRabbitConfig {
// 创建队列
@Bean
public Queue AMessage() {
return new Queue("fanout.A");
}
// 创建队列
@Bean
public Queue BMessage() {
return new Queue("fanout.B");
}
// 创建队列
@Bean
public Queue CMessage() {
return new Queue("fanout.C");
}
// 创建Fanout交换器
@Bean
FanoutExchange fanoutExchange() {
return new FanoutExchange("fanoutExchange");
}
// 将对列绑定到Fanout交换器
@Bean
Binding bindingExchangeA(Queue AMessage, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(AMessage).to(fanoutExchange);
}
// 将对列绑定到Fanout交换器
@Bean
Binding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(BMessage).to(fanoutExchange);
}
// 将对列绑定到Fanout交换器
@Bean
Binding bindingExchangeC(Queue CMessage, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(CMessage).to(fanoutExchange);
}
}
2.发送者
// producer
@Component
public class FanoutSender {
@Autowired
private AmqpTemplate rabbitTemplate;
public void send() {
String context = "hi, fanout msg ";
System.out.println("Sender : " + context);
this.rabbitTemplate.convertAndSend("fanoutExchange","", context);
}
}
//这里使用了A、B、C三个队列绑定到Fanout交换机上面，发送端的routing_key写任何字符都会被忽略。
// consumer
@RabbitHandler
public void process(String message) {
System.out.println("fanout Receiver A : " + message);
}
......
// test
@RunWith(SpringRunner.class)
@SpringBootTest
public class Test {
@Autowired
private FanoutSender fanoutSender;
@org.junit.Test
public void sendFanout () {
fanoutSender.send();
}
}
// result
fanout Receiver B: hi, fanout msg
fanout Receiver A : hi, fanout msg
fanout Receiver C: hi, fanout msg 以上是RabbitMQ的基本配置和使用，还需要大家继续深入学习理解相关知识。
如有披露或问题欢迎留言或者入群探讨
最后
以上就是笨笨长颈鹿为你收集整理的消息中间件（六）之-springBoot集成RabbitMQ的全部内容，希望文章能够帮你解决消息中间件（六）之-springBoot集成RabbitMQ所遇到的程序开发问题。
如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。
发表评论 取消回复