锐客网

  1. 首页 > 生活百科 > it技术 > >

SpringBoot集成RabbitMQ

IT知识springboot

一箫一剑平生意，负尽狂名十五年。这篇文章主要讲述SpringBoot集成RabbitMQ相关的知识，希望能为你提供帮助。

主页:??小王叔叔的博客??欢迎来访
支持:点赞?收藏?关注
RabbitMQ 即一个消息队列，主要是用来实现应用程序的异步和解耦，同时也能起到消息缓冲，消息分发的作用。
1.RabbitMQ介绍RabbitMQ是实现AMQP（高级消息队列协议）的消息中间件的一种，最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。RabbitMQ主要是为了实现系统之间的双向解耦而实现的。当生产者大量产生数据时，消费者无法快速消费，那么需要一个中间层。保存这个数据。
AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。
RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写，支持多种客户端，如：python、Ruby、.NET、java、JMS、C、php、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。
?
2.相关概念通常我们谈到队列服务, 会有三个概念： 发消息者、队列、收消息者，RabbitMQ 在这个基本概念之上, 多做了一层抽象, 在发消息者和 队列之间, 加入了交换器 (Exchange). 这样发消息者和队列就没有直接联系, 转而变成发消息者把消息给交换器, 交换器根据调度策略再把消息再给队列
  • 左侧 P 代表 生产者，也就是往 RabbitMQ 发消息的程序。
  • 中间即是 RabbitMQ，其中包括了 交换机 和 队列。
  • 右侧 C 代表 消费者，也就是往 RabbitMQ 拿消息的程序。
其中比较重要概念有 4 个，分别为：虚拟主机，交换机，队列，和绑定。
  • 虚拟主机：一个虚拟主机持有一组交换机、队列和绑定。为什么需要多个虚拟主机呢？很简单，RabbitMQ当中，用户只能在虚拟主机的粒度进行权限控制。  因此，如果需要禁止A组访问B组的交换机/队列/绑定，必须为A和B分别创建一个虚拟主机。每一个RabbitMQ服务器都有一个默认的虚拟主机“/”。
  • 交换机：Exchange 用于转发消息，但是它不会做存储  ，如果没有 Queue bind 到 Exchange 的话，它会直接丢弃掉 Producer 发送过来的消息。
这里有一个比较重要的概念：路由键  。消息到交换机的时候，交互机会转发到对应的队列中，那么究竟转发到哪个队列，就要根据该路由键。
  • 绑定：也就是交换机需要和队列相绑定，这其中如上图所示，是多对多的关系。
SpringBoot集成RabbitMQ非常简单，如果只是简单的使用配置非常少，springboot提供了spring-boot-starter-amqp项目对消息各种支持。
?

3.简单使用1.配置pom包
    主要是添加spring-boot-starter-amqp的支持
org.springframework.boot
spring-boot-starter-amqp


2.配置文件
配置rabbitmq的安装地址、端口以及账户信息.
spring.application.name=spirng-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");




4.发送者
rabbitTemplate是springboot 提供的默认实现
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);




5.接收者
@Component
@RabbitListener(queues = "hello")
public class HelloReceiver

@RabbitHandler
public void process(String hello)
System.out.println("Receiver: " + hello);




6.测试
@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitMqHelloTest

@Autowired
private HelloSender helloSender;

@Test
public void hello() throws Exception
helloSender.send();




注意，发送者和接收者的queue name必须一致，不然不能接收
多对多使用
一个发送者，N个接收者或者N个发送者和N个接收者会出现什么情况呢？
一对多发送
  对上面的代码进行了小改造,接收端注册了两个Receiver,Receiver1和Receiver2，发送端加入参数计数，接收端打印接收到的参数，下面是测试代码，发送一百条消息，来观察两个接收端的执行效果.
@Test
public void oneToMany() throws Exception
for (int i=0; i< 100; i++)
neoSender.send(i);


结果如下：
Receiver 1: spirng boot neo queue ****** 11
Receiver 2: spirng boot neo queue ****** 12
Receiver 2: spirng boot neo queue ****** 14
Receiver 1: spirng boot neo queue ****** 13
Receiver 2: spirng boot neo queue ****** 15
Receiver 1: spirng boot neo queue ****** 16
Receiver 1: spirng boot neo queue ****** 18
Receiver 2: spirng boot neo queue ****** 17
Receiver 2: spirng boot neo queue ****** 19
Receiver 1: spirng boot neo queue ****** 20


根据返回结果得到以下结论
一个发送者，N个接受者,经过测试会均匀的将消息发送到N个接收者中
【SpringBoot集成RabbitMQ】多对多发送
复制了一份发送者，加入标记，在一百个循环中相互交替发送
@Test
public void manyToMany() throws Exception
for (int i=0; i< 100; i++)
neoSender.send(i);
neoSender2.send(i);


结果如下：
Receiver 1: spirng boot neo queue ****** 20
Receiver 2: spirng boot neo queue ****** 20
Receiver 1: spirng boot neo queue ****** 21
Receiver 2: spirng boot neo queue ****** 21
Receiver 1: spirng boot neo queue ****** 22
Receiver 2: spirng boot neo queue ****** 22
Receiver 1: spirng boot neo queue ****** 23
Receiver 2: spirng boot neo queue ****** 23
Receiver 1: spirng boot neo queue ****** 24
Receiver 2: spirng boot neo queue ****** 24
Receiver 1: spirng boot neo queue ****** 25
Receiver 2: spirng boot neo queue ****** 25


结论：和一对多一样，接收端仍然会均匀接收到消息.
4.高级使用
对象的支持
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: Username=neo, pass=123456
Receiver object : Username=neo, pass=123456


Topic Exchange
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
Binding 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监听到消息
Fanout Exchange
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交换机上面的队列都收到了消息.?
??注意  ~

    推荐阅读


    上一篇：-save-dev 与 -save的区别

    下一篇：Sorry, name can no longer contain capital letters.