mq_mqtt - CSDN
mq 订阅
MQ(Message Queue)消息队列,是基础数据结构中“先进先出”的一种数据结构。一般用来解决应用解耦,异步消息,流量削锋等问题,实现高性能,高可用,可伸缩和最终一致性架构。 展开全文
MQ(Message Queue)消息队列,是基础数据结构中“先进先出”的一种数据结构。一般用来解决应用解耦,异步消息,流量削锋等问题,实现高性能,高可用,可伸缩和最终一致性架构。
信息
提    供
强大、 安全、 稳定的消息数据传递
外文名
Message Queue
典型应用
解耦、削峰、异步。
常见产品
RabbitMQ,RocketMQ,ActiveMQ,ZeroMQ,Kafka,IBM WebSphere
中文名
消息队列
MQMQ 介绍
MQ(Message Queue)消息队列,是基础数据结构中“先进先出”的一种数据机构。指把要传输的数据(消息)放在队列中,用队列机制来实现消息传递——生产者产生消息并把消息放入队列,然后由消费者去处理。消费者可以到指定队列拉取消息,或者订阅相应的队列,由MQ服务端给其推送消息。 [1]  MQ的作用 消息队列中间件是分布式系统中重要的组件,主要解决应用解耦,异步消息,流量削锋等问题,实现高性能,高可用,可伸缩和最终一致性架构。 [2]  解耦:一个业务需要多个模块共同实现,或者一条消息有多个系统需要对应处理,只需要主业务完成以后,发送一条MQ,其余模块消费MQ消息,即可实现业务,降低模块之间的耦合。异步:主业务执行结束后从属业务通过MQ,异步执行,减低业务的响应时间,提高用户体验。削峰:高并发情况下,业务异步处理,提供高峰期业务处理能力,避免系统瘫痪。MQ的缺点 [3]  1、系统可用性降低。依赖服务也多,服务越容易挂掉。需要考虑MQ瘫痪的情况2、系统复杂性提高。需要考虑消息丢失、消息重复消费、消息传递的顺序性3、业务一致性。主业务和从属业务一致性的处理
收起全文
精华内容
参与话题
  • 什么是MQ?什么时候使用MQ?为什么要使用MQ

    万次阅读 多人点赞 2018-04-20 00:01:56
    只是代码的搬运工~~~~~问题1:什么是MQ?什么时候使用MQ?为什么要使用MQMQ,是一种跨进程的通信机制,用于上下游传递消息。在互联网架构中,MQ是一种非常常见的上下游“逻辑解耦+物理解耦”的消息通信服务。使用...
    只是代码的搬运工~~~~~
    问题1:什么是MQ?什么时候使用MQ?为什么要使用MQ?
    MQ,是一种跨进程的通信机制,用于上下游传递消息。
    在互联网架构中,MQ是一种非常常见的上下游“逻辑解耦+物理解耦”的消息通信服务
    使用了MQ之后,消息发送上游只需要依赖MQ,不用依赖其他服务。
    问题1:什么是MQ?什么时候使用MQ?为什么要使用MQ?
    MQ,是一种跨进程的通信机制,用于上下游传递消息。
    在互联网架构中,MQ是一种非常常见的上下游“逻辑解耦+物理解耦”的消息通信服务
    使用了MQ之后,消息发送上游只需要依赖MQ,不用依赖其他服务。

    问题2:什么时候使用MQ

    【典型场景一:数据驱动的任务依赖】

     什么是任务依赖,举个栗子,互联网公司经常在凌晨进行一些数据统计任务,这些任务之间有一定的依赖关系,比如:

    1task3需要使用task2的输出作为输入

    2task2需要使用task1的输出作为输入

    这样的话,tast1, task2, task3之间就有任务依赖关系,必须task1先执行,再task2执行,载task3执行。


    对于这类需求,常见的实现方式是,使用cron人工排执行时间表

    1task10:00执行,经验执行时间为50分钟

    2task21:00执行(为task1预留10分钟buffer),经验执行时间也是50分钟

    3task32:00执行(为task2预留10分钟buffer

     

    这种方法的坏处是:

    1)如果有一个任务执行时间超过了预留buffer的时间,将会得到错误的结果,因为后置任务不清楚前置任务是否执行成功,此时要手动重跑任务,还有可能要调整排班表

    2)总任务的执行时间很长,总是要预留很多buffer,如果前置任务提前完成,后置任务不会提前开始

    3)如果一个任务被多个任务依赖,这个任务将会称为关键路径,排班表很难体现依赖关系,容易出错

    4)如果有一个任务的执行时间要调整,将会有多个任务的执行时间要调整


    无论如何,采用“cron排班表”的方法,各任务耦合,谁用过谁痛谁知道

     


    优化方案是,采用MQ解耦:

    1task1准时开始,结束后发一个“task1 done”的消息

    2task2订阅“task1 done”的消息,收到消息后第一时间启动执行,结束后发一个“task2 done”的消息

    3task3同理

     

    采用MQ优点是:

    1)不需要预留buffer,上游任务执行完,下游任务总会在第一时间被执行

    2)依赖多个任务,被多个任务依赖都很好处理,只需要订阅相关消息即可

    3)有任务执行时间变化,下游任务都不需要调整执行时间

     

    需要特别说明的是,MQ只用来传递上游任务执行完成的消息,并不用于传递真正的输入输出数据

     

    【典型场景二:上游不关心执行结果】

    上游需要关注执行结果时要用“调用”,上游不关注执行结果时,就可以使用MQ了。

     

    举个栗子58同城的很多下游需要关注“用户发布帖子”这个事件,比如招聘用户发布帖子后,招聘业务要奖励58豆,房产用户发布帖子后,房产业务要送2个置顶,二手用户发布帖子后,二手业务要修改用户统计数据。

     


    对于这类需求,常见的实现方式是,使用调用关系

    帖子发布服务执行完成之后,调用下游招聘业务、房产业务、二手业务,来完成消息的通知,但事实上,这个通知是否正常正确的执行,帖子发布服务根本不关注。

     

    这种方法的坏处是:

    1)帖子发布流程的执行时间增加

    2)下游服务当机,可能导致帖子发布服务受影响,上下游逻辑+物理依赖严重

    3)每当增加一个需要知道“帖子发布成功”信息的下游,修改代码的是帖子发布服务,这一点是最恶心的,属于架构设计中典型的依赖倒转谁用过谁痛谁知道(采用此法的请评论留言)

     


    优化方案是,采用MQ解耦:

    1)帖子发布成功后,向MQ发一个消息

    2)哪个下游关注“帖子发布成功”的消息,主动去MQ订阅

     

    采用MQ优点是:

    1)上游执行时间短

    2上下游逻辑+物理解耦,除了与MQ有物理连接,模块之间都不相互依赖

    3新增一个下游消息关注方,上游不需要修改任何代码

     

    典型场景三:上游关注执行结果,但执行时间很长】

     有时候上游需要关注执行结果,但执行结果时间很长(典型的是调用离线处理,或者跨公网调用),也经常使用回调网关+MQ来解耦。

     

    举个栗子,微信支付,跨公网调用微信的接口,执行时间会比较长,但调用方又非常关注执行结果,此时一般怎么玩呢?


    一般采用“回调网关+MQ”方案来解耦:

    1)调用方直接跨公网调用微信接口

    2)微信返回调用成功,此时并不代表返回成功

    3)微信执行完成后,回调统一网关

    4)网关将返回结果通知MQ

    5)请求方收到结果通知

     

    这里需要注意的是,不应该由回调网关来调用上游来通知结果,如果是这样的话,每次新增调用方,回调网关都需要修改代码,仍然会反向依赖,使用回调网关+MQ的方案,新增任何对微信支付的调用,都不需要修改代码啦。




    问题3:什么时候不使用MQ?
    既然MQ是互联网分层架构中的解耦利器,那所有通讯都使用MQ岂不是很好?这是一个严重的误区,调用与被调用的关系,是无法被MQ取代的。

    问题4:MQ不足?
    1系统更复杂,多了一个MQ组件
    2消息传递路径更长,延时会增加
    3)消息可靠性和重复性互为矛盾,消息不丢不重难以同时保证
    4上游无法知道下游的执行结果,这一点是很致命的

    举个栗子用户登录场景,登录页面调用passport服务,passport服务的执行结果直接影响登录结果,此处的“登录页面”与“passport服务”就必须使用调用关系,而不能使用MQ通信。
     
    无论如何,记住这个结论调用方实时依赖执行结果的业务场景,请使用调用,而不是MQ

    展开全文
  • 消息队列MQ的使用流程

    万次阅读 多人点赞 2018-09-19 19:21:21
    在大型平台的分布式项目中,消息队列MQ具有重要的作用,经常用在边缘业务功能的处理中,比如日志管理【下面将以Bug日志保存为例】,因为像日志保存、新用户注册发送邮件等操作都不是主干业务,可以放在消息队列异步...

    人工智能,零基础入门!http://www.captainbed.net/inner

    一、简介

    在大型平台的分布式项目中,消息队列MQ具有重要的作用,经常用在边缘业务功能的处理中,比如日志管理【下面将以Bug日志保存为例】,因为像日志保存、新用户注册发送邮件等操作都不是主干业务,可以放在消息队列异步处理,这样可以减小项目的阻塞和压力。下面介绍的是比较常用的一个消息队列ActiveMQ

    二、项目的一般业务逻辑

    就以Bug日志保存为例,介绍一般的业务逻辑:【如下图代码示例】

    【1】写一个接口,

    【2】然后调用注入的Service的业务逻辑方法,

    【3】接口返回结果,流程结束。

    三、要用消息队列情况的业务逻辑

    【1】首先要有另外一个新的MQ项目服务器,用于接收消息,和处理消息,暴露接口出来即可,

    【2】在原先的项目业务上修改:请求MQ项目暴露出来的接口,把数据传过去,如果是传输的数据是对象得注意这个对象进行序列话【就是实现Serializable接口】

    【3】需要注意的是,原先的项目中的service的业务逻辑代码要copy到MQ项目中,到这里你可能就明白了MQ的实现原理:原先的项目就负责发送消息给MQ项目,然后MQ项目上有一套完全一样的业务处理代码,MQ项目在拿到消息后再慢慢处理消息【异步进行】,说通俗一点,就相当于将业务转移到另外一个项目。

    四、MQ项目消息端配置 

    【1】原先项目中的service方法要copy过来,还有对应的实体类也要copy一致,

    【2】写一个接收消息的接口:

    【3】生产消息的生产端【生产者】

    【4】消费消息的消费端 【消费者】

    【5】配置queue的名字

     这是简单的运用,用起来还是蛮简单的,但是要了解里面的原理还要多多学习源码。

    展开全文
  • 消息中间件(一)MQ详解及四大MQ比较

    万次阅读 多人点赞 2018-08-29 22:05:58
    一、消息中间件相关知识 1、概述 消息队列已经逐渐成为企业IT系统内部通信的核心手段。它具有低耦合、可靠投递、广播、流量控制、最终一致性等一系列功能,成为异步RPC的主要手段之一。当今市面上有很多主流的消息...

    一、消息中间件相关知识

    1、概述

    消息队列已经逐渐成为企业IT系统内部通信的核心手段。它具有低耦合、可靠投递、广播、流量控制、最终一致性等一系列功能,成为异步RPC的主要手段之一。当今市面上有很多主流的消息中间件,如老牌的ActiveMQ、RabbitMQ,炙手可热的Kafka,阿里巴巴自主开发RocketMQ等。

     

    2、消息中间件的组成

          2.1 Broker

    消息服务器,作为server提供消息核心服务

          2.2 Producer

    消息生产者,业务的发起方,负责生产消息传输给broker,

          2.3 Consumer

    消息消费者,业务的处理方,负责从broker获取消息并进行业务逻辑处理

          2.4 Topic

    主题,发布订阅模式下的消息统一汇集地,不同生产者向topic发送消息,由MQ服务器分发到不同的订阅者,实现消息的       广播

          2.5 Queue

    队列,PTP模式下,特定生产者向特定queue发送消息,消费者订阅特定的queue完成指定消息的接收

          2.6 Message

    消息体,根据不同通信协议定义的固定格式进行编码的数据包,来封装业务数据,实现消息的传输

     

    3 消息中间件模式分类

          3.1 点对点

    PTP点对点:使用queue作为通信载体 

    说明: 
    消息生产者生产消息发送到queue中,然后消息消费者从queue中取出并且消费消息。 
    消息被消费以后,queue中不再存储,所以消息消费者不可能消费到已经被消费的消息。 Queue支持存在多个消费者,但是对一个消息而言,只会有一个消费者可以消费。

    3.2 发布/订阅

    Pub/Sub发布订阅(广播):使用topic作为通信载体 

    说明: 
    消息生产者(发布)将消息发布到topic中,同时有多个消息消费者(订阅)消费该消息。和点对点方式不同,发布到topic的消息会被所有订阅者消费。

    queue实现了负载均衡,将producer生产的消息发送到消息队列中,由多个消费者消费。但一个消息只能被一个消费者接受,当没有消费者可用时,这个消息会被保存直到有一个可用的消费者。 
    topic实现了发布和订阅,当你发布一个消息,所有订阅这个topic的服务都能得到这个消息,所以从1到N个订阅者都能得到一个消息的拷贝。

     

    4 消息中间件的优势

          4.1 系统解耦

    交互系统之间没有直接的调用关系,只是通过消息传输,故系统侵入性不强,耦合度低。

          4.2 提高系统响应时间

    例如原来的一套逻辑,完成支付可能涉及先修改订单状态、计算会员积分、通知物流配送几个逻辑才能完成;通过MQ架构设计,就可将紧急重要(需要立刻响应)的业务放到该调用方法中,响应要求不高的使用消息队列,放到MQ队列中,供消费者处理。

          4.3 为大数据处理架构提供服务

    通过消息作为整合,大数据的背景下,消息队列还与实时处理架构整合,为数据处理提供性能支持。

          4.4 Java消息服务——JMS

    Java消息服务(Java Message Service,JMS)应用程序接口是一个Java平台中关于面向消息中间件(MOM)的API,用于在两个应用程序之间,或分布式系统中发送消息,进行异步通信。 
    JMS中的P2P和Pub/Sub消息模式:点对点(point to point, queue)与发布订阅(publish/subscribe,topic)最初是由JMS定义的。这两种模式主要区别或解决的问题就是发送到队列的消息能否重复消费(多订阅)。

     

    5 消息中间件应用场景

           5.1 异步通信

    有些业务不想也不需要立即处理消息。消息队列提供了异步处理机制,允许用户把一个消息放入队列,但并不立即处理它。想向队列中放入多少消息就放多少,然后在需要的时候再去处理它们。

          5.2 解耦

    降低工程间的强依赖程度,针对异构系统进行适配。在项目启动之初来预测将来项目会碰到什么需求,是极其困难的。通过消息系统在处理过程中间插入了一个隐含的、基于数据的接口层,两边的处理过程都要实现这一接口,当应用发生变化时,可以独立的扩展或修改两边的处理过程,只要确保它们遵守同样的接口约束。

          5.3 冗余

    有些情况下,处理数据的过程会失败。除非数据被持久化,否则将造成丢失。消息队列把数据进行持久化直到它们已经被完全处理,通过这一方式规避了数据丢失风险。许多消息队列所采用的”插入-获取-删除”范式中,在把一个消息从队列中删除之前,需要你的处理系统明确的指出该消息已经被处理完毕,从而确保你的数据被安全的保存直到你使用完毕。

          5.4 扩展性

    因为消息队列解耦了你的处理过程,所以增大消息入队和处理的频率是很容易的,只要另外增加处理过程即可。不需要改变代码、不需要调节参数。便于分布式扩容。

          5.5 过载保护

    在访问量剧增的情况下,应用仍然需要继续发挥作用,但是这样的突发流量无法提取预知;如果以为了能处理这类瞬间峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力,而不会因为突发的超负荷的请求而完全崩溃。

          5.6 可恢复性

    系统的一部分组件失效时,不会影响到整个系统。消息队列降低了进程间的耦合度,所以即使一个处理消息的进程挂掉,加入队列中的消息仍然可以在系统恢复后被处理。

          5.7 顺序保证

    在大多使用场景下,数据处理的顺序都很重要。大部分消息队列本来就是排序的,并且能保证数据会按照特定的顺序来处理。

          5.8 缓冲

    在任何重要的系统中,都会有需要不同的处理时间的元素。消息队列通过一个缓冲层来帮助任务最高效率的执行,该缓冲有助于控制和优化数据流经过系统的速度。以调节系统响应时间。

          5.9 数据流处理

    分布式系统产生的海量数据流,如:业务日志、监控数据、用户行为等,针对这些数据流进行实时或批量采集汇总,然后进行大数据分析是当前互联网的必备技术,通过消息队列完成此类数据收集是最好的选择。

     

    6 消息中间件常用协议

          6.1 AMQP协议

    AMQP即Advanced Message Queuing Protocol,一个提供统一消息服务的应用层标准高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息,并不受客户端/中间件不同产品,不同开发语言等条件的限制。 
    优点:可靠、通用

          6.2 MQTT协议

    MQTT(Message Queuing Telemetry Transport,消息队列遥测传输)是IBM开发的一个即时通讯协议,有可能成为物联网的重要组成部分。该协议支持所有平台,几乎可以把所有联网物品和外部连接起来,被用来当做传感器和致动器(比如通过Twitter让房屋联网)的通信协议。 
    优点:格式简洁、占用带宽小、移动端通信、PUSH、嵌入式系统

          6.3 STOMP协议

    STOMP(Streaming Text Orientated Message Protocol)是流文本定向消息协议,是一种为MOM(Message Oriented Middleware,面向消息的中间件)设计的简单文本协议。STOMP提供一个可互操作的连接格式,允许客户端与任意STOMP消息代理(Broker)进行交互。 
    优点:命令模式(非topic\queue模式)

          6.4 XMPP协议

    XMPP(可扩展消息处理现场协议,Extensible Messaging and Presence Protocol)是基于可扩展标记语言(XML)的协议,多用于即时消息(IM)以及在线现场探测。适用于服务器之间的准即时操作。核心是基于XML流传输,这个协议可能最终允许因特网用户向因特网上的其他任何人发送即时消息,即使其操作系统和浏览器不同。 
    优点:通用公开、兼容性强、可扩展、安全性高,但XML编码格式占用带宽大

          6.5 其他基于TCP/IP自定义的协议

    有些特殊框架(如:redis、kafka、zeroMq等)根据自身需要未严格遵循MQ规范,而是基于TCP\IP自行封装了一套协议,通过网络socket接口进行传输,实现了MQ的功能。

     

    7 常见消息中间件MQ介绍

          7.1 RocketMQ

    阿里系下开源的一款分布式、队列模型的消息中间件,原名Metaq,3.0版本名称改为RocketMQ,是阿里参照kafka设计思想使用java实现的一套mq。同时将阿里系内部多款mq产品(Notify、metaq)进行整合,只维护核心功能,去除了所有其他运行时依赖,保证核心功能最简化,在此基础上配合阿里上述其他开源产品实现不同场景下mq的架构,目前主要多用于订单交易系统。

    具有以下特点:

    • 能够保证严格的消息顺序
    • 提供针对消息的过滤功能
    • 提供丰富的消息拉取模式
    • 高效的订阅者水平扩展能力
    • 实时的消息订阅机制
    • 亿级消息堆积能力

    官方提供了一些不同于kafka的对比差异: 
    https://rocketmq.apache.org/docs/motivation/

          7.2 RabbitMQ

    使用Erlang编写的一个开源的消息队列,本身支持很多的协议:AMQP,XMPP, SMTP,STOMP,也正是如此,使的它变的非常重量级,更适合于企业级的开发。同时实现了Broker架构,核心思想是生产者不会将消息直接发送给队列,消息在发送给客户端时先在中心队列排队。对路由(Routing),负载均衡(Load balance)、数据持久化都有很好的支持。多用于进行企业级的ESB整合。

          7.3 ActiveMQ

    Apache下的一个子项目。使用Java完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现,少量代码就可以高效地实现高级应用场景。可插拔的传输协议支持,比如:in-VM, TCP, SSL, NIO, UDP, multicast, JGroups and JXTA transports。RabbitMQ、ZeroMQ、ActiveMQ均支持常用的多种语言客户端 C++、Java、.Net,、Python、 Php、 Ruby等。

          7.4 Redis

    使用C语言开发的一个Key-Value的NoSQL数据库,开发维护很活跃,虽然它是一个Key-Value数据库存储系统,但它本身支持MQ功能,所以完全可以当做一个轻量级的队列服务来使用。对于RabbitMQ和Redis的入队和出队操作,各执行100万次,每10万次记录一次执行时间。测试数据分为128Bytes、512Bytes、1K和10K四个不同大小的数据。实验表明:入队时,当数据比较小时Redis的性能要高于RabbitMQ,而如果数据大小超过了10K,Redis则慢的无法忍受;出队时,无论数据大小,Redis都表现出非常好的性能,而RabbitMQ的出队性能则远低于Redis。

          7.5 Kafka

    Apache下的一个子项目,使用scala实现的一个高性能分布式Publish/Subscribe消息队列系统,具有以下特性:

    • 快速持久化:通过磁盘顺序读写与零拷贝机制,可以在O(1)的系统开销下进行消息持久化;
    • 高吞吐:在一台普通的服务器上既可以达到10W/s的吞吐速率;
    • 高堆积:支持topic下消费者较长时间离线,消息堆积量大;
    • 完全的分布式系统:Broker、Producer、Consumer都原生自动支持分布式,依赖zookeeper自动实现复杂均衡;
    • 支持Hadoop数据并行加载:对于像Hadoop的一样的日志数据和离线分析系统,但又要求实时处理的限制,这是一个可行的解决方案。

          7.6 ZeroMQ

    号称最快的消息队列系统,专门为高吞吐量/低延迟的场景开发,在金融界的应用中经常使用,偏重于实时数据通信场景。ZMQ能够实现RabbitMQ不擅长的高级/复杂的队列,但是开发人员需要自己组合多种技术框架,开发成本高。因此ZeroMQ具有一个独特的非中间件的模式,更像一个socket library,你不需要安装和运行一个消息服务器或中间件,因为你的应用程序本身就是使用ZeroMQ API完成逻辑服务的角色。但是ZeroMQ仅提供非持久性的队列,如果down机,数据将会丢失。如:Twitter的Storm中使用ZeroMQ作为数据流的传输。

    ZeroMQ套接字是与传输层无关的:ZeroMQ套接字对所有传输层协议定义了统一的API接口。默认支持 进程内(inproc) ,进程间(IPC) ,多播,TCP协议,在不同的协议之间切换只要简单的改变连接字符串的前缀。可以在任何时候以最小的代价从进程间的本地通信切换到分布式下的TCP通信。ZeroMQ在背后处理连接建立,断开和重连逻辑。

    特性:

    • 无锁的队列模型:对于跨线程间的交互(用户端和session)之间的数据交换通道pipe,采用无锁的队列算法CAS;在pipe的两端注册有异步事件,在读或者写消息到pipe的时,会自动触发读写事件。
    • 批量处理的算法:对于批量的消息,进行了适应性的优化,可以批量的接收和发送消息。
    • 多核下的线程绑定,无须CPU切换:区别于传统的多线程并发模式,信号量或者临界区,zeroMQ充分利用多核的优势,每个核绑定运行一个工作者线程,避免多线程之间的CPU切换开销。

     

    二、主要消息中间件的比较

     

    综合选择RabbitMq 

    展开全文
  • RabbitMQ教程-万字长文详解RabbitMQ

    万次阅读 多人点赞 2018-08-05 23:14:56
    RabbitMQ教程。RabbitMQ详解。消息队列(MQ),本质是个队列,队列中存放的内容是message。MQ用于不同进程Process/线程Thread之间通信。本文介绍RabbitMQ的使用。RabbitMQ实战教程。

    仅需一次订阅RabbitMQ,作者所有专栏都能看

    推荐【SpringBoothttps://blog.csdn.net/hellozpc/article/details/107095951
    推荐【SpringCloudhttps://blog.csdn.net/hellozpc/article/details/83692496
    推荐【Mybatishttps://blog.csdn.net/hellozpc/article/details/80878563

    **欢迎关注**
    **扫一扫**

    RabbitMQ实战教程

    1.什么是MQ

    • 消息队列(Message Queue,简称MQ),从字面意思上看,本质是个队列,FIFO先入先出,只不过队列中存放的内容是message而已。
      其主要用途:不同进程Process/线程Thread之间通信。

    为什么会产生消息队列?有几个原因:

    • 不同进程(process)之间传递消息时,两个进程之间耦合程度过高,改动一个进程,引发必须修改另一个进程,为了隔离这两个进程,在两进程间抽离出一层(一个模块),所有两进程之间传递的消息,都必须通过消息队列来传递,单独修改某一个进程,不会影响另一个;

    • 不同进程(process)之间传递消息时,为了实现标准化,将消息的格式规范化了,并且,某一个进程接受的消息太多,一下子无法处理完,并且也有先后顺序,必须对收到的消息进行排队,因此诞生了事实上的消息队列;

    • 关于消息队列的详细介绍请参阅:
      《Java帝国之消息队列》
      《一个故事告诉你什么是消息队列》
      《到底什么时候该使用MQ》

    • MQ框架非常之多,比较流行的有RabbitMq、ActiveMq、ZeroMq、kafka,以及阿里开源的RocketMQ。本文主要介绍RabbitMq

    • 本教程pdf及代码下载地址
      代码:https://download.csdn.net/download/zpcandzhj/10585077
      教程:https://download.csdn.net/download/zpcandzhj/10585092

    2.RabbitMQ

    2.1.RabbitMQ的简介

    这里写图片描述
    开发语言:Erlang – 面向并发的编程语言。

    这里写图片描述

    2.1.1.AMQP
    AMQP是消息队列的一个协议。

    这里写图片描述

    2.2.官网

    这里写图片描述

    2.3.MQ的其他产品

    这里写图片描述

    2.4.学习5种队列

    这里写图片描述

    2.5.安装文档

    这里写图片描述

    3.搭建RabbitMQ环境

    3.1.下载

    下载地址:http://www.rabbitmq.com/download.html

    3.2.windows下安装

    3.2.1.安装Erlang
    下载:http://www.erlang.org/download/otp_win64_17.3.exe
    安装:
    这里写图片描述
    这里写图片描述
    这里写图片描述
    这里写图片描述
    这里写图片描述
    安装完成。

    3.2.2.安装RabbitMQ
    这里写图片描述
    这里写图片描述
    这里写图片描述
    安装完成。

    开始菜单里出现如下选项:
    这里写图片描述

    启动、停止、重新安装等。

    3.2.3.启用管理工具
    1、双击这里写图片描述
    2、进入C:\Program Files (x86)\RabbitMQ Server\rabbitmq_server-3.4.1\sbin输入命令:
    rabbitmq-plugins enable rabbitmq_management
    这里写图片描述

    这样就启动了管理工具,可以试一下命令:
    停止:net stop RabbitMQ
    启动:net start RabbitMQ

    3、在浏览器中输入地址查看:http://127.0.0.1:15672/
    这里写图片描述
    4、使用默认账号登录:guest/ guest

    3.3.Linux下安装

    3.3.1.安装Erlang
    3.3.2.添加yum支持
    cd /usr/local/src/
    mkdir rabbitmq
    cd rabbitmq

    wget http://packages.erlang-solutions.com/erlang-solutions-1.0-1.noarch.rpm
    rpm -Uvh erlang-solutions-1.0-1.noarch.rpm

    rpm --import http://packages.erlang-solutions.com/rpm/erlang_solutions.asc

    使用yum安装:
    sudo yum install erlang
    这里写图片描述

    3.3.3.安装RabbitMQ
    上传rabbitmq-server-3.4.1-1.noarch.rpm文件到/usr/local/src/rabbitmq/
    安装:
    rpm -ivh rabbitmq-server-3.4.1-1.noarch.rpm

    3.3.4.启动、停止
    service rabbitmq-server start
    service rabbitmq-server stop
    service rabbitmq-server restart
    3.3.5.设置开机启动
    chkconfig rabbitmq-server on
    3.3.6.设置配置文件
    cd /etc/rabbitmq
    cp /usr/share/doc/rabbitmq-server-3.4.1/rabbitmq.config.example /etc/rabbitmq/

    mv rabbitmq.config.example rabbitmq.config
    3.3.7.开启用户远程访问
    vi /etc/rabbitmq/rabbitmq.config
    这里写图片描述
    注意要去掉后面的逗号。
    3.3.8.开启web界面管理工具
    rabbitmq-plugins enable rabbitmq_management
    service rabbitmq-server restart
    3.3.9.防火墙开放15672端口
    /sbin/iptables -I INPUT -p tcp --dport 15672 -j ACCEPT
    /etc/rc.d/init.d/iptables save

    3.4.安装的注意事项

    1、推荐使用默认的安装路径
    2、系统用户名必须是英文
    Win10改名字非常麻烦,具体方法百度
    这里写图片描述
    3、计算机名必须是英文
    这里写图片描述
    4、系统的用户必须是管理员

    如果安装失败应该如何解决:
    1、重装系统 – 不推荐
    2、将RabbitMQ安装到linux虚拟机中
    a)推荐
    3、使用别人安装好的RabbitMQ服务
    a)只要给你开通一个账户即可。
    b)使用公用的RabbitMQ服务,在192.168.50.22
    c)推荐

    常见错误:
    这里写图片描述

    3.5.安装完成后操作

    1、系统服务中有RabbitMQ服务,停止、启动、重启
    这里写图片描述
    2、打开命令行工具
    这里写图片描述
    如果找不到命令行工具,直接cd到相应目录:
    这里写图片描述
    输入命令rabbitmq-plugins enable rabbitmq_management启用管理插件
    这里写图片描述
    查看管理页面
    这里写图片描述
    通过默认账户 guest/guest 登录
    如果能够登录,说明安装成功。
    这里写图片描述

    4.添加用户

    4.1.添加admin用户

    这里写图片描述

    4.2.用户角色

    1、超级管理员(administrator)
    可登陆管理控制台,可查看所有的信息,并且可以对用户,策略(policy)进行操作。
    2、监控者(monitoring)
    可登陆管理控制台,同时可以查看rabbitmq节点的相关信息(进程数,内存使用情况,磁盘使用情况等)
    3、策略制定者(policymaker)
    可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分)。
    4、普通管理者(management)
    仅可登陆管理控制台,无法看到节点信息,也无法对策略进行管理。
    5、其他
    无法登陆管理控制台,通常就是普通的生产者和消费者。

    4.3.创建Virtual Hosts

    这里写图片描述

    选中Admin用户,设置权限:
    这里写图片描述
    看到权限已加:
    这里写图片描述

    4.4.管理界面中的功能

    这里写图片描述

    这里写图片描述

    5.学习五种队列

    这里写图片描述

    5.1.导入my-rabbitmq项目

    项目下载地址:
    https://download.csdn.net/download/zpcandzhj/10585077
    这里写图片描述

    5.2.简单队列

    5.2.1.图示
    这里写图片描述

    P:消息的生产者
    C:消息的消费者
    红色:队列

    生产者将消息发送到队列,消费者从队列中获取消息。
    5.2.2.导入RabbitMQ的客户端依赖

    <dependency>
       <groupId>com.rabbitmq</groupId>
       <artifactId>amqp-client</artifactId>
       <version>3.4.1</version>
    </dependency>
    

    5.2.3.获取MQ的连接

    package com.zpc.rabbitmq.util;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.Connection;
    
    public class ConnectionUtil {
    
        public static Connection getConnection() throws Exception {
            //定义连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            //设置服务地址
            factory.setHost("localhost");
            //端口
            factory.setPort(5672);
            //设置账号信息,用户名、密码、vhost
            factory.setVirtualHost("testhost");
            factory.setUsername("admin");
            factory.setPassword("admin");
            // 通过工程获取连接
            Connection connection = factory.newConnection();
            return connection;
        }
    }
    

    5.2.4.生产者发送消息到队列

    package com.zpc.rabbitmq.simple;
    
    import com.zpc.rabbitmq.util.ConnectionUtil;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    public class Send {
    
        private final static String QUEUE_NAME = "q_test_01";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            // 从连接中创建通道
            Channel channel = connection.createChannel();
    
            // 声明(创建)队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 消息内容
            String message = "Hello World!";
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");
            //关闭通道和连接
            channel.close();
            connection.close();
        }
    }
    

    5.2.5.管理工具中查看消息
    这里写图片描述

    点击上面的队列名称,查询具体的队列中的信息:
    这里写图片描述
    5.2.6.消费者从队列中获取消息

    package com.zpc.rabbitmq.simple;
    
    import com.zpc.rabbitmq.util.ConnectionUtil;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    public class Recv {
    
        private final static String QUEUE_NAME = "q_test_01";
    
        public static void main(String[] argv) throws Exception {
    
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            // 从连接中创建通道
            Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
    
            // 监听队列
            channel.basicConsume(QUEUE_NAME, true, consumer);
    
            // 获取消息
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" [x] Received '" + message + "'");
            }
        }
    }
    

    5.3.Work模式

    这里写图片描述

    5.3.1.图示
    这里写图片描述

    一个生产者、2个消费者。

    一个消息只能被一个消费者获取。
    5.3.2.消费者1

    package com.zpc.rabbitmq.work;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    import com.zpc.rabbitmq.util.ConnectionUtil;
    
    public class Recv {
    
        private final static String QUEUE_NAME = "test_queue_work";
    
        public static void main(String[] argv) throws Exception {
    
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 同一时刻服务器只会发一条消息给消费者
            //channel.basicQos(1);
    
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 监听队列,false表示手动返回完成状态,true表示自动
            channel.basicConsume(QUEUE_NAME, true, consumer);
    
            // 获取消息
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" [y] Received '" + message + "'");
                //休眠
                Thread.sleep(10);
                // 返回确认状态,注释掉表示使用自动确认模式
                //channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }
    

    5.3.3.消费者2

    package com.zpc.rabbitmq.work;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    import com.zpc.rabbitmq.util.ConnectionUtil;
    
    public class Recv2 {
    
        private final static String QUEUE_NAME = "test_queue_work";
    
        public static void main(String[] argv) throws Exception {
    
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 同一时刻服务器只会发一条消息给消费者
            //channel.basicQos(1);
    
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 监听队列,false表示手动返回完成状态,true表示自动
            channel.basicConsume(QUEUE_NAME, true, consumer);
    
            // 获取消息
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" [x] Received '" + message + "'");
                // 休眠1秒
                Thread.sleep(1000);
                //下面这行注释掉表示使用自动确认模式
                //channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }
    

    5.3.4.生产者
    向队列中发送100条消息。

    package com.zpc.rabbitmq.work;
    
    import com.zpc.rabbitmq.util.ConnectionUtil;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    public class Send {
    
        private final static String QUEUE_NAME = "test_queue_work";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            for (int i = 0; i < 100; i++) {
                // 消息内容
                String message = "" + i;
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
                System.out.println(" [x] Sent '" + message + "'");
    
                Thread.sleep(i * 10);
            }
    
            channel.close();
            connection.close();
        }
    }
    

    5.3.5.测试
    测试结果:
    1、消费者1和消费者2获取到的消息内容是不同的,同一个消息只能被一个消费者获取。
    2、消费者1和消费者2获取到的消息的数量是相同的,一个是消费奇数号消息,一个是偶数。

    • 其实,这样是不合理的,因为消费者1线程停顿的时间短。应该是消费者1要比消费者2获取到的消息多才对。
      RabbitMQ 默认将消息顺序发送给下一个消费者,这样,每个消费者会得到相同数量的消息。即轮询(round-robin)分发消息。

    • 怎样才能做到按照每个消费者的能力分配消息呢?联合使用 Qos 和 Acknowledge 就可以做到。
      basicQos 方法设置了当前信道最大预获取(prefetch)消息数量为1。消息从队列异步推送给消费者,消费者的 ack 也是异步发送给队列,从队列的视角去看,总是会有一批消息已推送但尚未获得 ack 确认,Qos 的 prefetchCount 参数就是用来限制这批未确认消息数量的。设为1时,队列只有在收到消费者发回的上一条消息 ack 确认后,才会向该消费者发送下一条消息。prefetchCount 的默认值为0,即没有限制,队列会将所有消息尽快发给消费者。

    • 2个概念

    • 轮询分发 :使用任务队列的优点之一就是可以轻易的并行工作。如果我们积压了好多工作,我们可以通过增加工作者(消费者)来解决这一问题,使得系统的伸缩性更加容易。在默认情况下,RabbitMQ将逐个发送消息到在序列中的下一个消费者(而不考虑每个任务的时长等等,且是提前一次性分配,并非一个一个分配)。平均每个消费者获得相同数量的消息。这种方式分发消息机制称为Round-Robin(轮询)。

    • 公平分发 :虽然上面的分配法方式也还行,但是有个问题就是:比如:现在有2个消费者,所有的奇数的消息都是繁忙的,而偶数则是轻松的。按照轮询的方式,奇数的任务交给了第一个消费者,所以一直在忙个不停。偶数的任务交给另一个消费者,则立即完成任务,然后闲得不行。而RabbitMQ则是不了解这些的。这是因为当消息进入队列,RabbitMQ就会分派消息。它不看消费者为应答的数目,只是盲目的将消息发给轮询指定的消费者。

    为了解决这个问题,我们使用basicQos( prefetchCount = 1)方法,来限制RabbitMQ只发不超过1条的消息给同一个消费者。当消息处理完毕后,有了反馈,才会进行第二次发送。
    还有一点需要注意,使用公平分发,必须关闭自动应答,改为手动应答。

    5.4.Work模式的“能者多劳”

    打开上述代码的注释:

    // 同一时刻服务器只会发一条消息给消费者
    channel.basicQos(1);
    
    //开启这行 表示使用手动确认模式
    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    

    同时改为手动确认:

    // 监听队列,false表示手动返回完成状态,true表示自动
    channel.basicConsume(QUEUE_NAME, false, consumer);
    

    测试:
    消费者1比消费者2获取的消息更多。

    5.5.消息的确认模式

    消费者从队列中获取消息,服务端如何知道消息已经被消费呢?

    模式1:自动确认
    只要消息从队列中获取,无论消费者获取到消息后是否成功消息,都认为是消息已经成功消费。
    模式2:手动确认
    消费者从队列中获取消息后,服务器会将该消息标记为不可用状态,等待消费者的反馈,如果消费者一直没有反馈,那么该消息将一直处于不可用状态。

    手动模式:
    这里写图片描述

    自动模式:
    这里写图片描述

    5.6.订阅模式

    这里写图片描述
    5.6.1.图示
    这里写图片描述

    解读:
    1、1个生产者,多个消费者
    2、每一个消费者都有自己的一个队列
    3、生产者没有将消息直接发送到队列,而是发送到了交换机
    4、每个队列都要绑定到交换机
    5、生产者发送的消息,经过交换机,到达队列,实现,一个消息被多个消费者获取的目的
    注意:一个消费者队列可以有多个消费者实例,只有其中一个消费者实例会消费
    这里写图片描述

    5.6.2.消息的生产者(看作是后台系统)
    向交换机中发送消息。

    package com.zpc.rabbitmq.subscribe;
    
    import com.zpc.rabbitmq.util.ConnectionUtil;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    public class Send {
    
        private final static String EXCHANGE_NAME = "test_exchange_fanout";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明exchange
            channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
    
            // 消息内容
            String message = "Hello World!";
            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");
    
            channel.close();
            connection.close();
        }
    }
    

    注意:消息发送到没有队列绑定的交换机时,消息将丢失,因为,交换机没有存储消息的能力,消息只能存在在队列中。
    5.6.3.消费者1(看作是前台系统)

    package com.zpc.rabbitmq.subscribe;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    import com.zpc.rabbitmq.util.ConnectionUtil;
    
    public class Recv {
    
        private final static String QUEUE_NAME = "test_queue_work1";
    
        private final static String EXCHANGE_NAME = "test_exchange_fanout";
    
        public static void main(String[] argv) throws Exception {
    
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定队列到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
    
            // 同一时刻服务器只会发一条消息给消费者
            channel.basicQos(1);
    
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 监听队列,手动返回完成
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
            // 获取消息
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" [Recv] Received '" + message + "'");
                Thread.sleep(10);
    
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }
    

    5.6.4.消费者2(看作是搜索系统)

    package com.zpc.rabbitmq.subscribe;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    import com.zpc.rabbitmq.util.ConnectionUtil;
    
    public class Recv2 {
    
        private final static String QUEUE_NAME = "test_queue_work2";
    
        private final static String EXCHANGE_NAME = "test_exchange_fanout";
    
        public static void main(String[] argv) throws Exception {
    
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定队列到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
    
            // 同一时刻服务器只会发一条消息给消费者
            channel.basicQos(1);
    
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 监听队列,手动返回完成
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
            // 获取消息
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" [Recv2] Received '" + message + "'");
                Thread.sleep(10);
    
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }
    

    5.6.5.测试
    测试结果:
    同一个消息被多个消费者获取。一个消费者队列可以有多个消费者实例,只有其中一个消费者实例会消费到消息。

    在管理工具中查看队列和交换机的绑定关系:

    这里写图片描述

    5.7.路由模式

    这里写图片描述
    5.7.1.图示
    这里写图片描述

    5.7.2.生产者
    这里写图片描述
    5.7.3.消费者1(假设是前台系统)
    这里写图片描述
    5.7.4.消费2(假设是搜索系统)
    这里写图片描述

    5.8.主题模式(通配符模式)

    这里写图片描述

    这里写图片描述

    5.8.1.图示
    这里写图片描述
    同一个消息被多个消费者获取。一个消费者队列可以有多个消费者实例,只有其中一个消费者实例会消费到消息。

    5.8.2.生产者

    package com.zpc.rabbitmq.topic;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    import com.zpc.rabbitmq.util.ConnectionUtil;
    
    public class Send {
    
        private final static String EXCHANGE_NAME = "test_exchange_topic";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明exchange
            channel.exchangeDeclare(EXCHANGE_NAME, "topic");
    
            // 消息内容
            String message = "Hello World!!";
            channel.basicPublish(EXCHANGE_NAME, "routekey.1", null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");
    
            channel.close();
            connection.close();
        }
    }
    

    5.8.3.消费者1(前台系统)

    package com.zpc.rabbitmq.topic;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    import com.zpc.rabbitmq.util.ConnectionUtil;
    
    public class Recv {
    
        private final static String QUEUE_NAME = "test_queue_topic_work_1";
    
        private final static String EXCHANGE_NAME = "test_exchange_topic";
    
        public static void main(String[] argv) throws Exception {
    
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定队列到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "routekey.*");
    
            // 同一时刻服务器只会发一条消息给消费者
            channel.basicQos(1);
    
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 监听队列,手动返回完成
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
            // 获取消息
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" [Recv_x] Received '" + message + "'");
                Thread.sleep(10);
    
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }
    

    5.8.4.消费者2(搜索系统)

    package com.zpc.rabbitmq.topic;
    
    import com.zpc.rabbitmq.util.ConnectionUtil;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    public class Recv2 {
    
        private final static String QUEUE_NAME = "test_queue_topic_work_2";
    
        private final static String EXCHANGE_NAME = "test_exchange_topic";
    
        public static void main(String[] argv) throws Exception {
    
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定队列到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "*.*");
    
            // 同一时刻服务器只会发一条消息给消费者
            channel.basicQos(1);
    
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 监听队列,手动返回完成
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
            // 获取消息
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" [Recv2_x] Received '" + message + "'");
                Thread.sleep(10);
    
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }
    

    6.Spring-Rabbit

    6.1.Spring项目

    http://spring.io/projects

    这里写图片描述

    6.2.简介

    这里写图片描述
    这里写图片描述

    6.3.使用

    6.3.1.消费者

    package com.zpc.rabbitmq.spring;
    
    /**
     * 消费者
     *
     * @author Evan
     */
    public class Foo {
    
        //具体执行业务的方法
        public void listen(String foo) {
            System.out.println("\n消费者: " + foo + "\n");
        }
    }
    

    6.3.2.生产者

    package com.zpc.rabbitmq.spring;
    
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.context.support.AbstractApplicationContext;
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    
    public class SpringMain {
        public static void main(final String... args) throws Exception {
            AbstractApplicationContext ctx = new ClassPathXmlApplicationContext(
                    "classpath:spring/rabbitmq-context.xml");
            //RabbitMQ模板
            RabbitTemplate template = ctx.getBean(RabbitTemplate.class);
            //发送消息
            template.convertAndSend("Hello, 鸟鹏!");
            Thread.sleep(1000);// 休眠1秒
            ctx.destroy(); //容器销毁
        }
    }
    

    6.3.3.配置文件
    1、定义连接工厂

    <!-- 定义RabbitMQ的连接工厂 -->
    <rabbit:connection-factory id="connectionFactory"
       host="127.0.0.1" port="5672" username="admin" password="admin"
       virtual-host="testhost" />
    

    2、定义模板(可以指定交换机或队列)

    <rabbit:template id="amqpTemplate" connection-factory="connectionFactory" exchange="fanoutExchange" />
    

    3、定义队列、交换机、以及完成队列和交换机的绑定

    <!-- 定义队列,自动声明 -->
    <rabbit:queue name="zpcQueue" auto-declare="true"/>
    
    <!-- 定义交换器,把Q绑定到交换机,自动声明 -->
    <rabbit:fanout-exchange name="fanoutExchange" auto-declare="true">
       <rabbit:bindings>
          <rabbit:binding queue="zpcQueue"/>
       </rabbit:bindings>
    </rabbit:fanout-exchange>
    

    4、定义监听

    <rabbit:listener-container connection-factory="connectionFactory">
       <rabbit:listener ref="foo" method="listen" queue-names="zpcQueue" />
    </rabbit:listener-container>
    
    <bean id="foo" class="com.zpc.rabbitmq.spring.Foo" />
    

    5、定义管理,用于管理队列、交换机等:

    <!-- MQ的管理,包括队列、交换器等 -->
    <rabbit:admin connection-factory="connectionFactory" />
    

    完整配置文件rabbitmq-context.xml

    <beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="http://www.springframework.org/schema/rabbit
       http://www.springframework.org/schema/rabbit/spring-rabbit-1.4.xsd
       http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans-4.1.xsd">
    
       <!-- 定义RabbitMQ的连接工厂 -->
       <rabbit:connection-factory id="connectionFactory"
          host="127.0.0.1" port="5672" username="admin" password="admin"
          virtual-host="testhost" />
    
       <!-- 定义Rabbit模板,指定连接工厂以及定义exchange -->
       <rabbit:template id="amqpTemplate" connection-factory="connectionFactory" exchange="fanoutExchange" />
       <!-- <rabbit:template id="amqpTemplate" connection-factory="connectionFactory"
          exchange="fanoutExchange" routing-key="foo.bar" /> -->
    
       <!-- MQ的管理,包括队列、交换器等 -->
       <rabbit:admin connection-factory="connectionFactory" />
    
       <!-- 定义队列,自动声明 -->
       <rabbit:queue name="zpcQueue" auto-declare="true"/>
       
       <!-- 定义交换器,把Q绑定到交换机,自动声明 -->
       <rabbit:fanout-exchange name="fanoutExchange" auto-declare="true">
          <rabbit:bindings>
             <rabbit:binding queue="zpcQueue"/>
          </rabbit:bindings>
       </rabbit:fanout-exchange>
       
    <!--   <rabbit:topic-exchange name="myExchange">
          <rabbit:bindings>
             <rabbit:binding queue="myQueue" pattern="foo.*" />
          </rabbit:bindings>
       </rabbit:topic-exchange> -->
    
       <!-- 队列监听 -->
       <rabbit:listener-container connection-factory="connectionFactory">
          <rabbit:listener ref="foo" method="listen" queue-names="zpcQueue" />
       </rabbit:listener-container>
    
       <bean id="foo" class="com.zpc.rabbitmq.spring.Foo" />
    
    </beans>
    

    6.4.持久化交换机和队列

    这里写图片描述

    持久化:将交换机或队列的数据保存到磁盘,服务器宕机或重启之后依然存在。
    非持久化:将交换机或队列的数据保存到内存,服务器宕机或重启之后将不存在。

    非持久化的性能高于持久化。

    如何选择持久化?非持久化? – 看需求。

    欢迎关注公众号「程猿薇茑」

    7.Spring集成RabbitMQ一个完整案例

    创建三个系统A,B,C
    A作为生产者,B、C作为消费者(B,C作为web项目启动)
    项目下载地址:https://download.csdn.net/download/zpcandzhj/10585077

    7.1.在A系统中发送消息到交换机

    7.1.1.导入依赖

    <?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
       <modelVersion>4.0.0</modelVersion>
    
       <groupId>com.zpc</groupId>
       <artifactId>myrabbitA</artifactId>
       <version>0.0.1-SNAPSHOT</version>
       <packaging>jar</packaging>
       <name>myrabbit</name>
    
       <dependencies>
          <dependency>
             <groupId>org.springframework.amqp</groupId>
             <artifactId>spring-rabbit</artifactId>
             <version>1.4.0.RELEASE</version>
          </dependency>
    
          <dependency>
             <groupId>com.alibaba</groupId>
             <artifactId>fastjson</artifactId>
             <version>1.2.47</version>
          </dependency>
       </dependencies>
    </project>
    

    7.1.2.队列和交换机的绑定关系
    实现:
    1、在配置文件中将队列和交换机完成绑定
    2、可以在管理界面中完成绑定
    a)绑定关系如果发生变化,需要修改配置文件,并且服务需要重启
    b)管理更加灵活
    c)更容易对绑定关系的权限管理,流程管理
    本例选择第2种方式
    7.1.3.配置
    rabbitmq-context.xml

    <beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="http://www.springframework.org/schema/rabbit
       http://www.springframework.org/schema/rabbit/spring-rabbit-1.4.xsd
       http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans-4.1.xsd">
    
       <!-- 定义RabbitMQ的连接工厂 -->
       <rabbit:connection-factory id="connectionFactory"
          host="127.0.0.1" port="5672" username="admin" password="admin"
          virtual-host="testhost" />
    
       <!-- MQ的管理,包括队列、交换器等 -->
       <rabbit:admin connection-factory="connectionFactory" />
    
       <!-- 定义交换器,暂时不把Q绑定到交换机,在管理界面去绑定 -->
       <!--<rabbit:topic-exchange name="topicExchange" auto-declare="true" ></rabbit:topic-exchange>-->
       <rabbit:direct-exchange name="directExchange" auto-declare="true" ></rabbit:direct-exchange>
       <!--<rabbit:fanout-exchange name="fanoutExchange" auto-declare="true" ></rabbit:fanout-exchange>-->
    
       <!-- 定义Rabbit模板,指定连接工厂以及定义exchange(exchange要和上面的一致) -->
       <!--<rabbit:template id="amqpTemplate" connection-factory="connectionFactory" exchange="topicExchange" />-->
       <rabbit:template id="amqpTemplate" connection-factory="connectionFactory" exchange="directExchange" />
       <!--<rabbit:template id="amqpTemplate" connection-factory="connectionFactory" exchange="fanoutExchange" />-->
    </beans>
    

    7.1.4.消息内容
    方案:
    1、消息内容使用对象做json序列化发送
    a)数据大
    b)有些数据其他人是可能用不到的
    2、发送特定的业务字段,如id、操作类型

    7.1.5.实现
    生产者MsgSender.java:

    package com.zpc.myrabbit;
    
    import com.alibaba.fastjson.JSON;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.context.support.AbstractApplicationContext;
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    
    import java.text.SimpleDateFormat;
    import java.util.Date;
    import java.util.HashMap;
    import java.util.Map;
    
    
    /**
     * 消息生产者
     */
    public class MsgSender {
        public static void main(String[] args) throws Exception {
            AbstractApplicationContext ctx = new ClassPathXmlApplicationContext(
                    "classpath:spring/rabbitmq-context.xml");
            //RabbitMQ模板
            RabbitTemplate template = ctx.getBean(RabbitTemplate.class);
    
            String date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());//24小时制
            //发送消息
            Map<String, Object> msg = new HashMap<String, Object>();
            msg.put("type", "1");
            msg.put("date", date);
            template.convertAndSend("type2", JSON.toJSONString(msg));
            Thread.sleep(1000);// 休眠1秒
            ctx.destroy(); //容器销毁
        }
    }
    

    7.2.在B系统接收消息

    7.2.1.导入依赖

    <?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>com.zpc</groupId>
        <artifactId>myrabbitB</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <packaging>war</packaging>
    
        <name>myrabbit</name>
        <properties>
            <spring.version>4.1.3.RELEASE</spring.version>
            <fastjson.version>1.2.46</fastjson.version>
        </properties>
    
        <dependencies>
            <dependency>
                <groupId>com.rabbitmq</groupId>
                <artifactId>amqp-client</artifactId>
                <version>3.4.1</version>
            </dependency>
            <dependency>
                <groupId>org.springframework.amqp</groupId>
                <artifactId>spring-rabbit</artifactId>
                <version>1.4.0.RELEASE</version>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-webmvc</artifactId>
                <version>${spring.version}</version>
            </dependency>
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>fastjson</artifactId>
                <version>1.2.47</version>
            </dependency>
        </dependencies>
    
        <build>
            <finalName>${project.artifactId}</finalName>
            <plugins>
                <!-- web层需要配置Tomcat插件 -->
                <plugin>
                    <groupId>org.apache.tomcat.maven</groupId>
                    <artifactId>tomcat7-maven-plugin</artifactId>
                    <configuration>
                        <path>/testRabbit</path>
                        <uriEncoding>UTF-8</uriEncoding>
                        <port>8081</port>
                    </configuration>
                </plugin>
            </plugins>
        </build>
    </project>
    

    7.2.2.配置

    <beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="http://www.springframework.org/schema/rabbit
       http://www.springframework.org/schema/rabbit/spring-rabbit-1.4.xsd
       http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans-4.1.xsd">
    
       <!-- 定义RabbitMQ的连接工厂 -->
       <rabbit:connection-factory id="connectionFactory"
          host="127.0.0.1" port="5672" username="admin" password="admin"
          virtual-host="testhost" />
    
       <!-- MQ的管理,包括队列、交换器等 -->
       <rabbit:admin connection-factory="connectionFactory" />
    
       <!-- 定义B系统需要监听的队列,自动声明 -->
       <rabbit:queue name="q_topic_testB" auto-declare="true"/>
    
       <!-- 队列监听 -->
       <rabbit:listener-container connection-factory="connectionFactory">
          <rabbit:listener ref="myMQlistener" method="listen" queue-names="q_topic_testB" />
       </rabbit:listener-container>
    
       <bean id="myMQlistener" class="com.zpc.myrabbit.listener.Listener" />
    </beans>
    

    7.2.3.具体处理逻辑

    public class Listener {
        //具体执行业务的方法
        public void listen(String msg) {
            System.out.println("\n消费者B开始处理消息: " + msg + "\n");
        }
    }
    

    7.2.4.在界面管理工具中完成绑定关系
    选中定义好的交换机(exchange)
    这里写图片描述
    1)direct
    这里写图片描述
    2)fanout
    这里写图片描述
    3)topic
    这里写图片描述

    7.3.在C系统中接收消息

    (和B系统配置差不多,无非是Q名和Q对应的处理逻辑变了)

    7.3.1.配置

    <beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="http://www.springframework.org/schema/rabbit
       http://www.springframework.org/schema/rabbit/spring-rabbit-1.4.xsd
       http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans-4.1.xsd">
    
       <!-- 定义RabbitMQ的连接工厂 -->
       <rabbit:connection-factory id="connectionFactory"
          host="127.0.0.1" port="5672" username="admin" password="admin"
          virtual-host="testhost" />
    
       <!-- MQ的管理,包括队列、交换器等 -->
       <rabbit:admin connection-factory="connectionFactory" />
    
       <!-- 定义C系统需要监听的队列,自动声明 -->
       <rabbit:queue name="q_topic_testC" auto-declare="true"/>
    
       <!-- 队列监听 -->
       <rabbit:listener-container connection-factory="connectionFactory">
          <rabbit:listener ref="myMQlistener" method="listen" queue-names="q_topic_testC" />
       </rabbit:listener-container>
    
       <bean id="myMQlistener" class="com.zpc.myrabbit.listener.Listener" />
    </beans>
    

    7.3.2.处理业务逻辑

    public class Listener {
    
        //具体执行业务的方法
        public void listen(String msg) {
            System.out.println("\n消费者C开始处理消息: " + msg + "\n");
        }
    }
    

    7.3.3.在管理工具中绑定队列和交换机
    见7.2.4

    7.3.4.测试
    分别启动B,C两个web应用,然后运行A的MsgSender主方法发送消息,分别测试fanout、direct、topic三种类型

    8.Springboot集成RabbitMQ

    8.1.简单队列

    1、配置pom文件,主要是添加spring-boot-starter-amqp的支持

    <dependency>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    

    2、配置application.properties文件
    配置rabbitmq的安装地址、端口以及账户信息

    spring.application.name=spirng-boot-rabbitmq
    spring.rabbitmq.host=127.0.0.1
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=admin
    spring.rabbitmq.password=admin
    

    3、配置rabbitmq队列

    package com.zpc.rabbitmq;
    
    import org.springframework.amqp.core.Queue;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class RabbitConfig {
        @Bean
        public Queue queue() {
            return new Queue("q_hello");
        }
    }
    

    4、rabbitmq发送者

    package com.zpc.rabbitmq;
    
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import java.text.SimpleDateFormat;
    import java.util.Date;
    
    @Component
    public class HelloSender {
        @Autowired
        private AmqpTemplate rabbitTemplate;
    
        public void send() {
            String date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());//24小时制
            String context = "hello " + date;
            System.out.println("Sender : " + context);
            //简单对列的情况下routingKey即为Q名
            this.rabbitTemplate.convertAndSend("q_hello", context);
        }
    }
    

    5、rabbitmq接收者

    package com.zpc.rabbitmq;
    
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    @Component
    @RabbitListener(queues = "q_hello")
    public class HelloReceiver {
    
        @RabbitHandler
        public void process(String hello) {
            System.out.println("Receiver  : " + hello);
        }
    }
    

    6、测试

    package com.zpc.rabbitmq;
    
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.junit4.SpringRunner;
    
    @RunWith(SpringRunner.class)
    @SpringBootTest
    public class RabbitMqHelloTest {
    
        @Autowired
        private HelloSender helloSender;
    
        @Test
        public void hello() throws Exception {
            helloSender.send();
        }
    }
    

    8.2.多对多使用(Work模式)

    注册两个Receiver:

    package com.zpc.rabbitmq;
    
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    @Component
    @RabbitListener(queues = "q_hello")
    public class HelloReceiver2 {
    
        @RabbitHandler
        public void process(String hello) {
            System.out.println("Receiver2  : " + hello);
        }
    
    }
    
    @Test
    public void oneToMany() throws Exception {
        for (int i=0;i<100;i++){
            helloSender.send(i);
            Thread.sleep(300);
        }
    }
    
    public void send(int i) {
        String date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());//24小时制
        String context = "hello " + i + " " + date;
        System.out.println("Sender : " + context);
        //简单对列的情况下routingKey即为Q名
        this.rabbitTemplate.convertAndSend("q_hello", context);
    }
    

    8.3.Topic Exchange(主题模式)

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

    首先对topic规则配置,这里使用两个队列(消费者)来演示。
    1)配置队列,绑定交换机

    package com.zpc.rabbitmq.topic;
    
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.core.TopicExchange;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class TopicRabbitConfig {
    
        final static String message = "q_topic_message";
        final static String messages = "q_topic_messages";
    
        @Bean
        public Queue queueMessage() {
            return new Queue(TopicRabbitConfig.message);
        }
    
        @Bean
        public Queue queueMessages() {
            return new Queue(TopicRabbitConfig.messages);
        }
    
        /**
         * 声明一个Topic类型的交换机
         * @return
         */
        @Bean
        TopicExchange exchange() {
            return new TopicExchange("mybootexchange");
        }
    
        /**
         * 绑定Q到交换机,并且指定routingKey
         * @param queueMessage
         * @param exchange
         * @return
         */
        @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.#");
        }
    }
    

    2)创建2个消费者
    q_topic_message 和q_topic_messages

    package com.zpc.rabbitmq.topic;
    
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    @Component
    @RabbitListener(queues = "q_topic_message")
    public class Receiver1 {
    
        @RabbitHandler
        public void process(String hello) {
            System.out.println("Receiver1  : " + hello);
        }
    }
    
    package com.zpc.rabbitmq.topic;
    
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    @Component
    @RabbitListener(queues = "q_topic_messages")
    public class Receiver2 {
    
        @RabbitHandler
        public void process(String hello) {
            System.out.println("Receiver2 : " + hello);
        }
    }
    

    3)消息发送者(生产者)

    package com.zpc.rabbitmq.topic;
    
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    @Component
    public class MsgSender {
    
        @Autowired
        private AmqpTemplate rabbitTemplate;
    
        public void send1() {
            String context = "hi, i am message 1";
            System.out.println("Sender : " + context);
            this.rabbitTemplate.convertAndSend("mybootexchange", "topic.message", context);
        }
    
    
        public void send2() {
            String context = "hi, i am messages 2";
            System.out.println("Sender : " + context);
            this.rabbitTemplate.convertAndSend("mybootexchange", "topic.messages", context);
        }
    }
    

    send1方法会匹配到topic.#和topic.message,两个Receiver都可以收到消息,发送send2只有topic.#可以匹配所有只有Receiver2监听到消息。
    4)测试

    package com.zpc.rabbitmq.topic;
    
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.junit4.SpringRunner;
    
    @RunWith(SpringRunner.class)
    @SpringBootTest
    public class RabbitTopicTest {
    
        @Autowired
        private MsgSender msgSender;
    
        @Test
        public void send1() throws Exception {
            msgSender.send1();
        }
    
        @Test
        public void send2() throws Exception {
            msgSender.send2();
        }
    }
    

    8.4.Fanout Exchange(订阅模式)

    • Fanout 就是我们熟悉的广播模式或者订阅模式,给Fanout交换机发送消息,绑定了这个交换机的所有队列都收到这个消息。
      1)配置队列,绑定交换机
    package com.zpc.rabbitmq.fanout;
    
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.FanoutExchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class FanoutRabbitConfig {
    
        @Bean
        public Queue aMessage() {
            return new Queue("q_fanout_A");
        }
    
        @Bean
        public Queue bMessage() {
            return new Queue("q_fanout_B");
        }
    
        @Bean
        public Queue cMessage() {
            return new Queue("q_fanout_C");
        }
    
        @Bean
        FanoutExchange fanoutExchange() {
            return new FanoutExchange("mybootfanoutExchange");
        }
    
        @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);
        }
    }
    

    2)创建3个消费者

    package com.zpc.rabbitmq.fanout;
    
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    @Component
    @RabbitListener(queues = "q_fanout_A")
    public class ReceiverA {
    
        @RabbitHandler
        public void process(String hello) {
            System.out.println("AReceiver  : " + hello + "/n");
        }
    }
    
    package com.zpc.rabbitmq.fanout;
    
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    @Component
    @RabbitListener(queues = "q_fanout_B")
    public class ReceiverB {
    
        @RabbitHandler
        public void process(String hello) {
            System.out.println("BReceiver  : " + hello + "/n");
        }
    }
    
    package com.zpc.rabbitmq.fanout;
    
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    @Component
    @RabbitListener(queues = "q_fanout_C")
    public class ReceiverC {
    
        @RabbitHandler
        public void process(String hello) {
            System.out.println("CReceiver  : " + hello + "/n");
        }
    }
    

    3)生产者

    package com.zpc.rabbitmq.fanout;
    
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    @Component
    public class MsgSenderFanout {
    
        @Autowired
        private AmqpTemplate rabbitTemplate;
    
        public void send() {
            String context = "hi, fanout msg ";
            System.out.println("Sender : " + context);
            this.rabbitTemplate.convertAndSend("mybootfanoutExchange","", context);
        }
    }
    

    4)测试

    package com.zpc.rabbitmq.fanout;
    
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.junit4.SpringRunner;
    
    @RunWith(SpringRunner.class)
    @SpringBootTest
    public class RabbitFanoutTest {
    
        @Autowired
        private MsgSenderFanout msgSender;
    
        @Test
        public void send1() throws Exception {
            msgSender.send();
        }
    }
    

    结果如下,三个消费者都收到消息:
    AReceiver : hi, fanout msg
    CReceiver : hi, fanout msg
    BReceiver : hi, fanout msg

    9.总结

    推荐springCloud教程:
    https://blog.csdn.net/hellozpc/article/details/83692496

    推荐Springboot2.0教程:
    https://blog.csdn.net/hellozpc/article/details/82531834

    **欢迎关注公众号【程猿薇茑】**
    **微信扫一扫**
    展开全文
  • MQ详解及四大MQ比较

    万次阅读 2019-01-11 20:58:57
    一、消息中间件相关知识 1、概述 消息队列已经逐渐成为企业IT系统内部通信的核心手段。它具有低耦合、可靠投递、广播、流量控制、最终一致性等一系列功能,成为异步RPC的主要手段之一。当今市面上有很多主流的消息...
  • 不少候选人,说自己项目里用了 Redis、MQ,但是其实他并不知道自己为什么要用这个东西。其实说白了,就是为了用而用,或者是别人设计的架构,他从头到尾都没思考过。 没有对自己的架构问过为什么的人,一定是平时...
  • 为什么要使用MQ和到底什么时候要使用MQ

    万次阅读 多人点赞 2018-06-10 18:41:13
    原文地址:http://mp.weixin.qq.com/s/Brd-j3IcljcY7BV01r712Q一、缘起一切脱离业务的架构设计与新技术引入都是耍流氓。 引入一个技术之前,首先应该解答的... 最近分享了几篇MQ相关的文章:《MQ如何实现延时消息...
  • 消息队列 为什么写这篇文章? 博主有两位朋友分别是小A和小B: 小A,工作于传统软件行业(某社保局的软件外包公司),每天工作内容就是和产品聊聊需求,改改业务逻辑。再不然就是和运营聊聊天,写几个SQL,生成下报表...
  • mq的基本介绍和基本用法

    万次阅读 多人点赞 2018-08-04 11:54:41
    1.什么是MQ,有什么用? MQ 是message queue ,消息队列,也叫消息中间件,遵守JMS(java message service)规范的一种软件。(同时还有另一个叫AMQP的应用层协议,语言无关性不受产品 语言等限制,rabbitMQ支持这个 )...
  • 【消息队列MQ】各类MQ比较

    万次阅读 多人点赞 2012-09-11 21:22:24
    目前业界有很多MQ产品,我们作如下对比: RabbitMQ 是使用Erlang编写的一个开源的消息队列,本身支持很多的协议:AMQP,XMPP, SMTP, STOMP,也正是如此,使的它变的非常重量级,更适合于企业级的开发。同时实现了...
  • 消息中间件MQ与RabbitMQ面试题(2020最新版)

    万次阅读 多人点赞 2020-03-01 11:11:21
    文章目录为什么使用MQMQ的优点消息队列有什么优缺点?RabbitMQ有什么优缺点?你们公司生产环境用的是什么消息中间件?Kafka、ActiveMQ、RabbitMQ、RocketMQ 有什么优缺点?MQ 有哪些常见问题?如何解决这些问题?...
  • MQ

    千次阅读 2016-06-24 20:41:16
  • MQ消息队列详解 近期有了想跳槽的打算,所以自己想巩固一下自己的技术,想了解一些面试比较容易加分的项,近期准备深入研究一下redis和mq这两样,这总体上都是为了解决服务器并发的原因,刚翻到了一篇有关于mq的,...
  • 消息队列mq的原理及实现方法

    万次阅读 多人点赞 2016-07-18 20:50:09
    消息队列技术是分布式应用间交换信息的一种技术。消息队列可驻留在内存或磁盘上,队列存储消息直到它们被应用程序读走。通过消息队列,应用程序可独立地执行--它们不需要知道彼此的位置、或在继续执行前不需要等待...
  • 概述 消息队列是Linux IPC中很常用的一种通信方式,它通常用来在不同进程间发送特定格式的消息数据。 消息队列和之前讨论过的管道和FIFO有很大的区别,主要有以下两点: 一个进程向消息队列写入消息之前,并不需要...
  • MQ-2烟雾传感器的原理及使用教程

    万次阅读 2019-11-15 22:47:26
    一、MQ-2烟雾传感器简介 MQ-2常用于家庭和工厂的气体泄漏监测装置,适宜于液化气、苯、烷、酒精、氢气、烟雾等的探测。故因此,MQ-2可以准确来说是一个多种气体探测器。 MQ-2的探测范围极其的广泛。它的优点:灵敏...
  • MQ2烟雾传感器

    万次阅读 多人点赞 2019-04-09 10:59:56
    1、MQ-2气体传感器所使用的气敏材料是在清洁空气中电导率较低的二氧化锡(SnO2)。当传感器所处环境中存在可燃气体时,传感器的电导率随空气中可燃气体浓度的增加而增大。使用简单的电路即可将电导率的变化转换为与该...
  • 一、MQ简介及特点 MQ全称为Message Queue, 消息队列(MQ)是一种应用程序对应用程序的通信方法。应用程序通过写和检索出入列队的针对应用程序的数据(消息)来通信,而无需专用连接来链接它们。消息传递指的是...
  • MQ命令学习总结大全MQ常用命令

    千次阅读 2017-06-06 13:57:06
    一.MQ基本操作  MQ中有几个很重要的组件:队列管理器(QueueManager)、队列(Queue)和通道(Channel)。其基本的操作方法如下:  创建队列管理器  crtmqm –q QMgrName  -q是指创建缺省的队列管理器 ...
  • WebSphere® MQ uses two different types of channels: A message channel, which is a unidirectional communications link between two queue managers. WebSphere MQ uses message channels to transfer mes...
1 2 3 4 5 ... 20
收藏数 125,087
精华内容 50,034
关键字:

mq