当前位置: 首页 > news >正文

rabbitmq入门_SpringBoot整合RabbitMQ(一)快速入门

点击上方“Java知音”,选择“置顶公众号”

技术文章第一时间送达!

//本文作者:微笑面对生活

//欢迎加入Java知音写作,开源、分享!

MQ全称为Message Queue, 消息队列(MQ)是一种应用程序对应用程序的通信方法。MQ是消费-生产者模型的一个典型的代表,一端往消息队列中不断写入消息,而另一端则可以读取队列中的消息。

消息中间件最主要的作用是解耦,中间件最标准的用法是生产者生产消息传送到队列,消费者从队列中拿取消息并处理,生产者不用关心是谁来消费,消费者不用关心谁在生产消息,从而达到解耦的目的。在分布式的系统中,消息队列也会被用在很多其它的方面,比如:分布式事务的支持,RPC的调用等等。

RabbitMQ介绍

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

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

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

相关概念

通常我们谈到队列服务, 会有三个概念: 发消息者、队列、收消息者,RabbitMQ 在这个基本概念之上, 多做了一层抽象, 在发消息者和 队列之间, 加入了交换器 (Exchange). 这样发消息者和队列就没有直接联系, 转而变成发消息者把消息给交换器, 交换器根据调度策略再把消息再给队列。

那么,其中比较重要的概念有 4 个,分别为:虚拟主机,交换机,队列,和绑定。

  • 虚拟主机:一个虚拟主机持有一组交换机、队列和绑定。为什么需要多个虚拟主机呢?很简单,RabbitMQ当中,用户只能在虚拟主机的粒度进行权限控制。 因此,如果需要禁止A组访问B组的交换机/队列/绑定,必须为A和B分别创建一个虚拟主机。每一个RabbitMQ服务器都有一个默认的虚拟主机“/”。

  • 交换机:Exchange 用于转发消息,但是它不会做存储 ,如果没有 Queue bind 到 Exchange 的话,它会直接丢弃掉 Producer 发送过来的消息。 这里有一个比较重要的概念:路由键 。消息到交换机的时候,交互机会转发到对应的队列中,那么究竟转发到哪个队列,就要根据该路由键。

  • 绑定:也就是交换机需要和队列相绑定,这其中如上图所示,是多对多的关系。

四种交换机(Exchange)

交换机的功能主要是接收消息并且转发到绑定的队列,交换机不存储消息,在启用ack模式后,交换机找不到队列会返回错误。交换机有四种类型:Direct, topic, Headers and Fanout

1. Direct Exchange

direct 类型的行为是"先匹配, 再投送". 即在绑定时设定一个 routingkey, 消息的routingkey 匹配时, 才会被交换器投送到绑定的队列中去.是RabbitMQ默认的交换机模式,也是最简单的模式,根据key全文匹配去寻找队列。

配置:设置一个路由键

   public  static  final String QUEUE="queue";

   /**

    * direct 交换机模式

    */

   @Bean

   public Queue queue(){

       return new Queue(QUEUE,true);

   }

发送服务:

@Service

@Slf4j

public class MQSender {

   @Autowired

   AmqpTemplate amqpTemplate;

   public void send(Object message){

       String msg = (String) message;

       log.info("send msg"+message);

       amqpTemplate.convertAndSend(MQConfig.QUEUE,msg);

   }

}

接收服务:

@Service

@Slf4j

public class MQReceiver {

   //监听的queue

   @RabbitListener(queues = MQConfig.QUEUE)

   public void receive(String msg){

       log.info("receive msg "+msg);

   }

}

测试:

   @Autowired

   private MQSender sender;

   sender.send("hello direct Exchange");

7c5aebb1f701b7034bd7781b460089d2.png

2. Topic Exchange

按规则转发消息(最灵活) 转发消息主要是根据通配符。 在这种交换机下,队列和交换机的绑定会定义一种路由模式,那么,通配符就要在这种路由模式和路由键之间匹配后交换机才能转发消息。

路由键必须是一串字符,用句号(.) 隔开,

路由模式必须包含一个 星号(*),主要用于匹配路由键指定位置的一个单词, 井号(#)就表示相当于一个或者多个单词

配置类:

   public  static  final String TOPIC_QUEUE1="topic.queue1";

   public  static  final String TOPIC_QUEUE2="topic.queue2";

   public  static  final String ROUTING_KEY1="topic.key1";

   public  static  final String ROUTING_KEY2="topic.#";

   /**

    * Topic 交换机模式  可以用通配符

    */

   @Bean

   public Queue topicQueue1(){

       return new Queue(TOPIC_QUEUE1,true);

   }

   @Bean

   public Queue topicQueue2(){

       return new Queue(TOPIC_QUEUE2,true);

   }

   @Bean

   public TopicExchange topicExchange(){

       return new TopicExchange(TOPIC_EXCHANGE);

   }

   @Bean

   public Binding topicBinding1(){

       return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with(ROUTING_KEY1);

   }

   @Bean

   public Binding topicBinding2(){

       return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with(ROUTING_KEY2);

   }

发送类:

   public void sendTopic(Object message){

       String msg = (String) message;

       log.info("send topic message"+msg);

       amqpTemplate.convertAndSend(MQConfig.TOPIC_EXCHANGE,"topic.key1",msg+"1");

       amqpTemplate.convertAndSend(MQConfig.TOPIC_EXCHANGE,"topic.key2",msg+"2");

   }

接收类:

   @RabbitListener(queues = MQConfig.TOPIC_QUEUE1)

   public void receiveTopic1(String msg){

       log.info("receive topic1 msg "+msg);

   }

测试:

   @Autowired

   private MQSender sender;

   sender.sendTopic("hello topic Exchange");

9e6049ae58c86f188b36289f0f166a27.png

3. Headers Exchange

设置header attribute参数类型的交换机,相较于 direct 和 topic 固定地使用 routing_key , headers 则是一个自定义匹配规则的类型. 在队列与交换器绑定时, 会设定一组键值对规则, 消息中也包括一组键值对( headers 属性), 当这些键值对有一对, 或全部匹配时, 消息被投送到对应队列.

   public static final String HEADER_EXCHANGE="headerExchange";

   /**

    * Header 交换机模式

    */

   @Bean

   public HeadersExchange headersExchange(){

       return new HeadersExchange(HEADER_EXCHANGE);

   }

   @Bean

   public Queue headerQueue(){

       return new Queue(HEADER_QUEUE2,true);

   }

   // 绑定需要指定header,如果不匹配 则不能使用

   @Bean

   public Binding headerBinding(){

       Map<String,Object> map = new HashMap();

       map.put("header1","value1");

       map.put("header2","value2");

       return BindingBuilder.bind(headerQueue()).to(headersExchange()).whereAll(map).match();

   }

   public void sendHeader(Object massage){

       String msg = (String) massage;

       log.info("send fanout message: "+msg);

       MessageProperties properties = new MessageProperties();

       properties.setHeader("header1","value1");

       properties.setHeader("header2","value2");

       Message obj = new Message(msg.getBytes(),properties);

       amqpTemplate.convertAndSend(MQConfig.HEADER_EXCHANGE,"",obj);

   }

用MessageProperties来添加Header信息,然后与接收者的header比对。我都设置的是"header1","value1";"header2","value2"

   //监听 header模式的queue

   @RabbitListener(queues = MQConfig.HEADER_QUEUE2)

   //因为发送的是 byte 类型,所以接受也是该数据类型

   public void receiveHeader(byte[] message){

       log.info("header queue message"+new String(message));

   }

测试:

   @Autowired

   private MQSender sender;

   sender.sendHeader("hello header Exchange");

a0da5fcf3bc56ae49075a6886bd723be.png

4. Fanout Exchange

转发消息到所有绑定队列,消息广播的模式,不管路由键或者是路由模式,会把消息发给绑定给它的全部队列,如果配置了routing_key会被忽略。

   public static final String FANOUT_EXCHANGE="fanoutExchange";

   /**

    * Fanout 交换机模式(广播模式),不用绑定key

    */

   @Bean

   public FanoutExchange fanoutExchange(){

       return new FanoutExchange(FANOUT_EXCHANGE);

   }

   @Bean

   public Binding fanoutBinding1(){

       return BindingBuilder.bind(topicQueue1()).to(fanoutExchange());

   }

   @Bean

   public Binding fanoutBinding2(){

       return BindingBuilder.bind(topicQueue2()).to(fanoutExchange());

   }

   public void sendFanout(Object massage){

       String msg = (String) massage;

       log.info("send fanout message: "+msg);

       amqpTemplate.convertAndSend(MQConfig.FANOUT_EXCHANGE,"",msg);

   }

测试:

   @Autowired

   private MQSender sender;

   sender.sendFanout("hello fanout Exchange");

de32f06714f6bf7829f4b9d4321572b3.png

补充

这个示例是基于springboot的示例。

pom依赖

   org.springframework.boot

   spring-boot-starter-amqp

配置文件

 rabbitmq:

   host: 127.0.0.1

   port: 5672

   username: guest

   password: guest

   # 这个账号密码只能连接本地的mq,远程的话需要配置

   virtual-host: /

   listener:

     simple:

       concurrency: 10

       max-concurrency: 10

       prefetch: 1 # 从队列每次取一个

       auto-startup: true

       default-requeue-rejected: true # 失败后重试

后续会用它来实现一个小小的抢票功能。


更多Java技术文章,尽在【Java知音】网站。

网址:www.javazhiyin.com  ,搜索Java知音可达!

干货!必须好看☟

相关文章:

  • select count(*) from返回的类型_SQL面试题类型整理
  • 用自底向上算法为一组整数构造一个大根堆。_Mathematical Cryptography笔记:整数分解和RSA...
  • .xml 下拉列表_RecyclerView嵌套recyclerview实现二级下拉列表,包含自定义IOS对话框...
  • java怎么实现自动退出功能_教你用python操作Excel 轻松实现自动读写功能
  • python 语义网络_Python好书从入门到进阶整理好送你
  • cloud压缩怎么彻底删除_财务软件T+Cloud操作指南
  • python字符串反码_python中的进制转换和原码,反码,补码
  • python假设检验和区间估计_推断统计分析(三):python实现假设检验
  • c++ char 转int_程序员每日一题-int和char还可以一起玩耍
  • java double转int_Java中的换形师-数据类型转换
  • matplotlib 标签_[Matplotlib官方教程]使用指南 0.7
  • python微信群发消息_Python3 itchat实现微信定时发送群消息的实例代码
  • easyconnect无法在mac上使用_如何在Mac上使用pyenv运行Python的多个版本
  • isdebugenabled_日志框架中为什么有isDebugEnabled方法?
  • python hashlib_python import hashlib出现问题
  • php的引用
  • IE9 : DOM Exception: INVALID_CHARACTER_ERR (5)
  • bearychat的java client
  • Cookie 在前端中的实践
  • css的样式优先级
  • Golang-长连接-状态推送
  • java中具有继承关系的类及其对象初始化顺序
  • Joomla 2.x, 3.x useful code cheatsheet
  • leetcode98. Validate Binary Search Tree
  • RedisSerializer之JdkSerializationRedisSerializer分析
  • 浮现式设计
  • 回顾2016
  • 简单基于spring的redis配置(单机和集群模式)
  • 浅谈JavaScript的面向对象和它的封装、继承、多态
  • 提升用户体验的利器——使用Vue-Occupy实现占位效果
  • 用jquery写贪吃蛇
  • LevelDB 入门 —— 全面了解 LevelDB 的功能特性
  • # 数据结构
  • ###STL(标准模板库)
  • (003)SlickEdit Unity的补全
  • (C#)一个最简单的链表类
  • (libusb) usb口自动刷新
  • (ResultSet.TYPE_SCROLL_INSENSITIVE,ResultSet.CONCUR_READ_ONLY)讲解
  • (附源码)ssm航空客运订票系统 毕业设计 141612
  • (入门自用)--C++--抽象类--多态原理--虚表--1020
  • (十六)一篇文章学会Java的常用API
  • (一)硬件制作--从零开始自制linux掌上电脑(F1C200S) <嵌入式项目>
  • .net mvc部分视图
  • .NET MVC第五章、模型绑定获取表单数据
  • .Net mvc总结
  • .NET6 命令行启动及发布单个Exe文件
  • .NET导入Excel数据
  • .Net多线程总结
  • .net项目IIS、VS 附加进程调试
  • /usr/lib/mysql/plugin权限_给数据库增加密码策略遇到的权限问题
  • @reference注解_Dubbo配置参考手册之dubbo:reference
  • [ C++ ] STL_stack(栈)queue(队列)使用及其重要接口模拟实现
  • [20170728]oracle保留字.txt
  • [BT]BUUCTF刷题第8天(3.26)
  • [C++参考]拷贝构造函数的参数必须是引用类型