rabbitmq即一个消息队列,主要是拿来实现应用程序的异步和前馈,同时也能起到消息缓冲,消息分发的作用。
消息中间件在互联网公司的使用中越来越多,刚刚还听到新闻阿里将RocketMQ捐赠给了Apache,其实了明天的主角还是讲RabbitMQ。消息中间件最主要的作用是前馈,中间件最标准的用法是生产者生产消息传送到队列,消费者从队列中拿取消息并处理,生产者不用关心是谁来消费,消费者不用关心谁在生产消息,进而达到前馈的目的。在分布式的系统中,消息队列也会被用在好多其它的方面消息队列使用流程图,例如:分布式事务的支持,RPC的调用等等。
曾经仍然使用的是ActiveMQ,在实际的生产使用中也出现了一些小问题,在网路查阅了好多的资料后,决定尝试使用RabbitMQ来替换ActiveMQ,RabbitMQ的高可用性、高性能、灵活性等一些特性吸引了我们,查阅了一些资料整理出此文。
RabbitMQ介绍
RabbitMQ是实现AMQP(中级消息队列合同)的消息中间件的一种,最初起源于金融系统,用于在分布式系统中储存转发消息,在易用性、扩展性、高可用性等方面表现不俗。RabbitMQ主要是为了实现系统之间的单向前馈而实现的。当生产者大量形成数据时,消费者难以快速消费消息队列使用流程图,这么须要一个中间层。保存这个数据。
AMQP,即AdvancedmessageQueuingProtocol,中级消息队列合同,是应用层合同的一个开放标准,为面向消息的中间件设计。消息中间件主要用于组件之间的前馈,消息的发送者无需晓得消息使用者的存在,反之亦然。AMQP的主要特点是面向消息、队列、路由(包括点对点和发布/订阅)、可靠性、安全。
RabbitMQ是一个开源的AMQP实现,服务器端用Erlang语言编撰,支持多种顾客端,如:Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等,支持AJAX。用于在分布式系统中储存转发消息,在易用性、扩展性、高可用性等方面表现不俗。
相关概念
一般我们提到队列服务,会有三个概念:发消息者、队列、收消息者,RabbitMQ在这个基本概念之上,多做了一层具象,在发消息者和队列之间,加入了交换器(Exchange).这样发消息者和队列就没有直接联系,转而弄成发消息者把消息给交换器,交换器按照调度策略再把消息再给队列。
这么,其中比较重要的概念有4个,分别为:虚拟主机,交换机,队列,和绑定。
交换机(Exchange)
交换机的功能主要是接收消息但是转发到绑定的队列,交换机不储存消息,在启用ack模式后,交换机找不到队列会返回错误。交换机有四种类型:direct,topic,HeadersandFanout
DirectExchange
DirectExchange是RabbitMQ默认的交换机模式,也是最简单的模式,按照key全文匹配去找寻队列。
第一个X–Q1就有一个bindingkey,名子为orange;X–Q2就有2个bindingkey,名子为black和green。当消息中的路由键和这个bindingkey对应上的时侯,这么就晓得了该消息去到哪一个队列中。
Ps:为何X到Q2要有black,green,2个bindingkey呢,一个不就行了吗?–这个主要是由于可能又有Q3,而Q3只接受black的信息,而Q2除了接受black的信息,还接受green的信息。
TopicExchange
TopicExchange转发消息主要是依据转义。在这些交换机下,队列和交换机的绑定会定义一种路由模式,这么,键值就要在这些路由模式和路由键之间匹配后交换机能够转发消息。
在这些交换机模式下:
具体代码发送的时侯还是一样,第一个参数表示交换机,第二个参数表示routingkey,第三个参数即消息。如下:
rabbitTemplate.convertAndSend("testTopicExchange","key1.a.c.key2", " this is RabbitMQ!");
topic和direct类似,只是匹配上支持了”模式”,在”点分”的routing_key方式中,可以使用两个键值:
HeadersExchange
headers也是按照规则匹配,相较于direct和topic固定地使用routing_key,headers则是一个自定义匹配规则的类型.在队列与交换器绑定时,会设定一组通配符对规则,消息中也包括一组通配符对(headers属性),当这种通配符对有一对,或全部匹配时,消息被投送到对应队列.
FanoutExchange
FanoutExchange消息广播的模式,不管路由键或则是路由模式,会把消息发给绑定给它的全部队列,假如配置了routing_key会被忽视。
SpringBoot集成RabbitMQ
SpringBoot集成RabbitMQ十分简单,倘若只是简单的使用配置十分少,SpringBoot提供了spring-boot-starter-amqp项目对消息各类支持。
简单使用
1、配置Pom包,主要是添加spring-boot-starter-amqp的支持
org.springframework.boot
spring-boot-starter-amqp
2、配置文件
配置RabbitMQ的安装地址、端口以及帐户信息
spring. application .name=Spring-boot-rabbitmq
spring.rabbitmq.host=192.168.0.86
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123456
3、队列配置
@ Configuration
public class RabbitConfig {
@Bean
public Queue Queue() {
return new Queue("hello");
}
}
3、发送者
rabbitTemplate是SpringBoot提供的默认实现
@component
public class HelloSender {
@Autowired
private AmqpTemplate rabbitTemplate;
public void send() {
String context = "hello " + new Date();
System.out.println("Sender : " + context);
this.rabbitTemplate.convertAndSend("hello", context);
}
}
4、接收者
@Component
@RabbitListener(queues = "hello")
public class HelloReceiver {
@RabbitHandler
public void process(String hello) {
System.out.println("Receiver : " + hello);
}
}
5、测试
@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitMqHelloTest {
@Autowired
private HelloSender helloSender;
@Test
public void hello() throws Exception {
helloSender.send();
}
}
多对多使用
一个发送者,N个接收者或则N个发送者和N个接收者会出现哪些情况呢?
一对多发送
对里面的代码进行了小整修,接收端注册了两个Receiver,Receiver1和Receiver2,发送端加入参数计数,接收端复印接收到的参数,下边是测试代码,发送一百条消息,来观察两个接收端的执行疗效
@Test
public void oneToMany() throws Exception {
for (int i=0;i<100;i++){
neoSender.send(i);
}
}
结果如下:
Receiver 1: Spring boot neo queue ****** 11
Receiver 2: Spring boot neo queue ****** 12
Receiver 2: Spring boot neo queue ****** 14
Receiver 1: Spring boot neo queue ****** 13
Receiver 2: Spring boot neo queue ****** 15
Receiver 1: Spring boot neo queue ****** 16
Receiver 1: Spring boot neo queue ****** 18
Receiver 2: Spring boot neo queue ****** 17
Receiver 2: Spring boot neo queue ****** 19
Receiver 1: Spring boot neo queue ****** 20
按照返回结果得到以下推论
一个发送者,N个接受者,经过测试会均匀的将消息发送到N个接收者中
多对多发送
复制了一份发送者,加入标记,在一百个循环中互相交替发送
结果如下:
Receiver 1: Spring boot neo queue ****** 20
Receiver 2: Spring boot neo queue ****** 20
Receiver 1: Spring boot neo queue ****** 21
Receiver 2: Spring boot neo queue ****** 21
Receiver 1: Spring boot neo queue ****** 22
Receiver 2: Spring boot neo queue ****** 22
Receiver 1: Spring boot neo queue ****** 23
Receiver 2: Spring boot neo queue ****** 23
Receiver 1: Spring boot neo queue ****** 24
Receiver 2: Spring boot neo queue ****** 24
Receiver 1: Spring boot neo queue ****** 25
Receiver 2: Spring boot neo queue ****** 25
推论:和一对多一样,接收端依然会均匀接收到消息
中级使用
对象的支持
SpringBoot以及完美的支持对象的发送和接收,不须要愈发的配置。
//发送者
public void send(User user) {
System.out.println("Sender object: " + user.toString());
this.rabbitTemplate.convertAndSend("object", user);
}
...
//接收者
@RabbitHandler
public void process(User user) {
System.out.println("Receiver object : " + user);
}
结果如下:
Sender object: User{name='neo', pass='123456'}
Receiver object : User{name='neo', pass='123456'}
TopicExchange
topic是RabbitMQ中最灵活的一种方法,可以按照routing_key自由的绑定不同的队列
首先对topic规则配置,这儿使用两个队列来测试
@Configuration
public class TopicRabbitConfig {
final static String message = "topic.message";
final static String messages = "topic.messages";
@Bean
public Queue queueMessage() {
return new Queue(TopicRabbitConfig.message);
}
@Bean
public Queue queueMessages() {
return new Queue(TopicRabbitConfig.messages);
}
@Bean
TopicExchange exchange() {
return new TopicExchange("exchange");
}
@Bean
bind ing bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {
return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
}
@Bean
Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {
return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");
}
}
使用queueMessages同时匹配两个队列,queueMessage只匹配“topic.message”队列
public void send1() {
String context = "hi, i am message 1";
System.out.println("Sender : " + context);
this.rabbitTemplate.convertAndSend("exchange", "topic.message", context);
}
public void send2() {
String context = "hi, i am messages 2";
System.out.println("Sender : " + context);
this.rabbitTemplate.convertAndSend("exchange", "topic.messages", context);
}
发送send1会匹配到topic.#和topic.message两个Receiver都可以收到消息,发送send2只有topic.#可以匹配所有只有Receiver2窃听到消息
FanoutExchange
Fanout就是我们熟悉的广播模式或则订阅模式,给Fanout交换机发送消息,绑定了这个交换机的所有队列都收到这个消息。
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");
}
@Bean
FanoutExchange fanoutExchange() {
return new FanoutExchange("fanoutExchange");
}
@Bean
Binding bindingExchangeA(Queue AMessage,FanoutExchange fanoutExchange) {
return BindingBuilder.bind(AMessage).to(fanoutExchange);
}
@Bean
Binding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(BMessage).to(fanoutExchange);
}
@Bean
Binding bindingExchangeC(Queue CMessage, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(CMessage).to(fanoutExchange);
}
}
这儿使用了A、B、C三个队列绑定到Fanout交换机里面,发送端的routing_key写任何字符就会被忽视:
public void send() {
String context = "hi, fanout msg ";
System.out.println("Sender : " + context);
this.rabbitTemplate.convertAndSend("fanoutExchange","", context);
}
结果如下:
Sender : hi, fanout msg
...
fanout Receiver B: hi, fanout msg
fanout Receiver A : hi, fanout msg
fanout Receiver C: hi, fanout msg
结果说明,绑定到fanout交换机前面的队列都收到了消息
文章内容早已升级到SpringBoot2.x
暂无评论内容