消息队列是实现AMQP的异步和解耦而实现的

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

图片[1]-消息队列是实现AMQP的异步和解耦而实现的-唐朝资源网

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 {

图片[2]-消息队列是实现AMQP的异步和解耦而实现的-唐朝资源网

@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

图片[3]-消息队列是实现AMQP的异步和解耦而实现的-唐朝资源网

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

© 版权声明
THE END
喜欢就支持一下吧
点赞122 分享
评论 抢沙发
头像
欢迎您留下宝贵的见解!
提交
头像

昵称

取消
昵称表情代码图片