spring使用rabbitmq,springboot rabbitmq分布式事务

  spring使用rabbitmq,springboot rabbitmq分布式事务

  00-1010一、RabbitMQ介绍二。相关概念三。简单使用1。配置pom包2。配置文件3。队列配置4。发送者5。接收器6。测试四。高级使用1。话题交流2。扇出交换机

  

目录

RabbitMQ是一种实现AMQP(高级消息队列协议)的消息中间件。它起源于金融系统,用于在分布式系统中存储和转发消息。它在易用性、可扩展性和高可用性方面表现出色。RabbitMQ主要实现系统间的双向解耦。当生产者产生大量数据,而消费者又不能快速消费时,那么就需要一个中间层。保存这些数据。

 

  AMQP(高级消息队列协议)是一个开放的应用层协议标准,是为面向消息的中间件设计的。消息中间件主要用来解耦组件,消息的发送方不需要知道消息消费者的存在,反之亦然。AMQP的主要特点是面向消息、面向队列、路由(包括点对点和发布/订阅)、可靠性和安全性。

  RabbitMQ是AMQP的开源实现。服务器端是用Erlang语言编写的,支持多种客户端,比如Python,Ruby,NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等。并且支持AJAX。用于在分布式系统中存储和转发消息,在易用性、可扩展性和高可用性方面表现良好。

  00-1010通常,当我们谈到队列服务时,会有三个概念:发送者、队列和接收者。RabbitMQ在这个基本概念上做了额外的抽象,增加了发送者和队列之间的交换。这样,发送方和队列之间没有直接联系,而是由发送方将消息交给交换方,交换方再根据调度策略将消息交给队列。

  左边的p代表生产者,也就是向RabbitMQ发送消息的程序。中间是RabbitMQ,包括交换机和队列。右边的c代表消费者,也就是从RabbitMQ获取消息的程序。其中,重要的概念有 4 个有:虚拟主机、交换机、队列和绑定。

  虚拟主机:的虚拟主机拥有一组交换机、队列和绑定。为什么需要多个虚拟主机?很简单,在RabbitMQ中,用户只能在虚拟主机的粒度上控制自己的权限。因此,如果需要禁止A组访问B组的交换机/队列/绑定,则必须分别为A组和B组创建一个虚拟主机。每个RabbitMQ服务器都有一个默认的虚拟主机“/”。交换机:交换用于转发邮件,但不存储邮件。如果没有队列绑定到Exchange,它将直接丢弃Producer发送的消息。这里有一个重要的概念:路由键。当消息到达交换机时,交互机会将其转发到相应的队列,所以转发到哪个队列取决于路由键。

  绑定:意味着交换机需要绑定到队列,如上图所示,是多对多的关系。SpringBoot集成RabbitMQ非常简单。如果只是简单的配置,springboot在spring-boot-starter-amqp项目中提供了对消息的各种支持。

  

一、RabbitMQ介绍

 

  00-1010主要增加了对spring-boot-starter-amqp的支持。

  dependencygroupidorg . spring framework . boot/groupid artifactId spring-boot-starter-amqp/artifactId/dependency

  

二、相关概念

配置rabbitmq的安装地址、端口以及账户信息.

 

  spring . application . name=spir ng-boot-rabbit MQ spring . rabbit MQ . host=192 . 168 . 0 . 86 spring . rabbit MQ . port=5672 spring . rabbit MQ . username=admin spring . rabbit MQ . password=123456

  

三、简单使用

@配置公共

 

  class RabbitConfig {@Beanpublic Queue Queue() {return new Queue("hello");}}

 

  

4.发送者

rabbitTemplate是springboot 提供的默认实现public class HelloSender {@Autowiredprivate 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 {@RabbitHandlerpublic void process(String hello) {System.out.println("Receiver : " + hello);}}

 

  

6.测试

@RunWith(SpringRunner.class)@SpringBootTestpublic class RabbitMqHelloTest {@Autowiredprivate HelloSender helloSender;@Testpublic void hello() throws Exception {helloSender.send();}}
注意:发送者和接收者的queue name必须一致,不然不能接收

 

  

多对多使用

 

  一个发送者,N个接收者或者N个发送者和N个接收者会出现什么情况呢?

  一对多发送

   对上面的代码进行了小改造,接收端注册了两个Receiver,Receiver1和Receiver2,发送端加入参数计数,接收端打印接收到的参数,下面是测试代码,发送一百条消息,来观察两个接收端的执行效果.

  

@Testpublic void oneToMany() throws Exception {for (int i=0;i<100;i++){neoSender.send(i);}}

结果如下:

 

  

Receiver 1: spirng boot neo queue ****** 11Receiver 2: spirng boot neo queue ****** 12Receiver 2: spirng boot neo queue ****** 14Receiver 1: spirng boot neo queue ****** 13Receiver 2: spirng boot neo queue ****** 15Receiver 1: spirng boot neo queue ****** 16Receiver 1: spirng boot neo queue ****** 18Receiver 2: spirng boot neo queue ****** 17Receiver 2: spirng boot neo queue ****** 19Receiver 1: spirng boot neo queue ****** 20

 

  

根据返回结果得到以下结论

 

  一个发送者,N个接受者,经过测试会均匀的将消息发送到N个接收者中

  多对多发送

  复制了一份发送者,加入标记,在一百个循环中相互交替发送

  

@Testpublic void manyToMany() throws Exception {for (int i=0;i<100;i++){neoSender.send(i);neoSender2.send(i);}}

结果如下:

 

  

Receiver 1: spirng boot neo queue ****** 20Receiver 2: spirng boot neo queue ****** 20Receiver 1: spirng boot neo queue ****** 21Receiver 2: spirng boot neo queue ****** 21Receiver 1: spirng boot neo queue ****** 22Receiver 2: spirng boot neo queue ****** 22Receiver 1: spirng boot neo queue ****** 23Receiver 2: spirng boot neo queue ****** 23Receiver 1: spirng boot neo queue ****** 24Receiver 2: spirng boot neo queue ****** 24Receiver 1: spirng boot neo queue ****** 25Receiver 2: spirng boot neo queue ****** 25

 

  

结论:和一对多一样,接收端仍然会均匀接收到消息.

 

  

 

  

四、高级使用

//对象的支持//springboot以及完美的支持对象的发送和接收,不需要格外的配置。//发送者public void send(User user) {System.out.println("Sender object: " + user.toString());this.rabbitTemplate.convertAndSend("object", user);}...//接受者@RabbitHandlerpublic void process(User user) {System.out.println("Receiver object : " + user);}

结果如下:

 

  

Sender object: User{name='neo', pass='123456'}Receiver object : User{name='neo', pass='123456'}

 

  

 

  

1.Topic Exchange

topic 是RabbitMQ中最灵活的一种方式,可以根据routing_key自由的绑定不同的队列

 

  首先对topic规则配置,这里使用两个队列来测试

  

 

  

@Configurationpublic class TopicRabbitConfig {final static String message = "topic.message";final static String messages = "topic.messages";@Beanpublic Queue queueMessage() {return new Queue(TopicRabbitConfig.message);}@Beanpublic Queue queueMessages() {return new Queue(TopicRabbitConfig.messages);}@BeanTopicExchange exchange() {return new TopicExchange("exchange");}@BeanBinding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");}@BeanBinding 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监听到消息

 

  

 

  

2.Fanout Exchange

Fanout 就是我们熟悉的广播模式或者订阅模式,给Fanout交换机发送消息,绑定了这个交换机的所有队列都收到这个消息。

 

  

 

  Fanout 相关配置:

  

@Configurationpublic class FanoutRabbitConfig {@Beanpublic Queue AMessage() {return new Queue("fanout.A");}@Beanpublic Queue BMessage() {return new Queue("fanout.B");}@Beanpublic Queue CMessage() {return new Queue("fanout.C");}@BeanFanoutExchange fanoutExchange() {return new FanoutExchange("fanoutExchange");}@BeanBinding bindingExchangeA(Queue AMessage,FanoutExchange fanoutExchange) {return BindingBuilder.bind(AMessage).to(fanoutExchange);}@BeanBinding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange) {return BindingBuilder.bind(BMessage).to(fanoutExchange);}@BeanBinding 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 msgfanout Receiver A : hi, fanout msgfanout Receiver C: hi, fanout msg

 

  

结果说明,绑定到fanout交换机上面的队列都收到了消息.

 

  到此这篇关于SpringBoot集成RabbitMQ和概念介绍的文章就介绍到这了,更多相关SpringBoot集成RabbitMQ内容请搜索盛行IT以前的文章或继续浏览下面的相关文章希望大家以后多多支持盛行IT!

郑重声明:本文由网友发布,不代表盛行IT的观点,版权归原作者所有,仅为传播更多信息之目的,如有侵权请联系,我们将第一时间修改或删除,多谢。

留言与评论(共有 条评论)
   
验证码: