java-rabbitmq 中间件 RabbitMQ

RabbitMQ

一、什么是中间件？二、中间件技术及架构的概述三、基于消息中间件的分布式系统架构1.消息中间件的应用场景2.常见的消息中间件3.消息中间件的本质及设计4.消息中间件的核心组成部分

四、消息队列协议1. 什么是协议2. AMQP协议3. MQTT协议4. OpenMessage协议5. Kafka协议

五、消息队列持久化六、消息的分发策略七、高可用&高可靠1. 高可用2. 高可靠

八、 RabbitMQ入门及安装1. 安装RabbitMQ2. Web管理界面及授权3. Docker安装RabbitMQ

九、原理讲解1. AMQP协议2. 核心组成部分

十、 RabbitMQ模式1. 简单模式2. 四种 exchage1）fanout类型（发布与订阅模式）2）Direct类型（Routing模式）3）topic类型（主题模式）4）Headers类型 【几乎不用】

3. Work模式1）轮询模式2）公平模式

十一、使用场景十二、SpringBoot整合1. fanout类型2. direct类型3. topic类型

十三、RabbitMQ高级1. 过期时间TTL2. 死信队列

十四、运行监控1. 内存控制2. 磁盘预警3. 内存换页

十五、集群搭建十六、分布式事务概念十七、分布式事务实现1. 系统间调用de事务回滚问题2. 分布式事务整体设计思路3. 解决可靠生产问题-定时重发4. 可靠消费

十八、集群配置详解附录

感谢飞哥的分享：B站视频地址

参考文章：消息队列–RabbitMq

源码gitee地址：https://gitee.com/wxawangxinan/rabbit-mq.git

一、什么是中间件？

1、简介

中间件（Middleware）：是处于操作系统和应用程序之间的软件，也有人认为它应该属于操作系统中的一部分。人们在使用中间件时，往往是一组中间件集成在一起，构成一个平台（包括开发平台和运行平台），但在这组中间件中必须要有一个通信中间件，即 中间件 = 平台 + 通信。这个定义也限定了只有用于分布式系统中才能称为中间件，同时还可以把它与 支撑软件和实用软件区分开来。

2、举例

RMI（Remote Method Invocations, 远程调用）Load Balancing(负载均衡，将访问负荷分散到各个服务器中)Transparent Fail-over(透明的故障切换)Clustering(集群,用多个小的服务器代替大型机）Back-end-Integration(后端集成，用现有的、新开发的系统如何去集成遗留的系统)Transaction事务（全局/局部）全局事务（分布式事务）局部事务（在同一数据库联接内的事务）Dynamic Redeployment(动态重新部署,在不停止原系统的情况下，部署新的系统）System Management(系统管理)Threading(多线程处理)Message-oriented Middleware面向消息的中间件（异步的调用编程）Component Life Cycle(组件的生命周期管理)Resource pooling（资源池）Security（安全）Caching（缓存） 满足中间件特性的都可以称作中间件，例如:rabbitMQ,Nginx,MySQL,Redis

为什么使用消息中间件

具体地说，中间件屏蔽了底层操作系统的复杂性，使程序开发人员面对一个简单而统一的开发环境，减少程序设计的复杂性，将注意力集中在自己的业务上，不必再为程序在不同系统软件上的移植而重复工作，从而大大减少了技术上的负担。中间件带给应用系统的，不只是开发的简便、开发周期的缩短，也减少了系统的维护、运行和管理的工作量，还减少了计算机总体费用的投入。

中间件特点

为解决分布异构问题，人们提出了中间件(middleware)的概念。中间件是位于平台(硬件和操作系统)和应用之间的通用服务，如下图所示，这些服务具有标准的程序接口和协议。针对不同的操作系统和硬件平台，它们可以有符合接口和协议规范的多种实现。 很难给中间件一个严格的定义，但中间件应具有如下的一些特点： （1）满足大量应用的需要 （2）运行于多种硬件和OS平台 （3）支持分布计算，提供跨网络、硬件和OS平台的透明性的应用或服务的交互 （4）支持标准的协议 （5）支持标准的接口

由于标准接口对于可移植性和标准协议对于互操作性的重要性，中间件已成为许多标准化工作的主要部分。对于应用软件开发，中间件远比操作系统和网络服务更为重要，中间件提供的程序接口定义了一个相对稳定的高层应用环境，不管底层的计算机硬件和系统软件怎样更新换代，只要将中间件升级更新，并保持中间件对外的接口定义不变，应用软件几乎不需任何修改，从而保护了企业在应用软件开发和维护中的重大投资。

二、中间件技术及架构的概述

知识路线

学习技巧

1：理解中间件在项目架构中的作用，以及各中间件的底层实现。 2：可以使用一些类比的生活概念去理解中间件， 3：使用一些流程图或者脑图的方式去梳理各个中间件在架构中的作用 4：尝试用java技术去实现中间件的原理 5：静下来去思考中间件在项目中设计的和使用的原因

三、基于消息中间件的分布式系统架构

从上图中看，消息中间件的作用是： 1：利用可靠的消息传递机制，进行系统和系统直接的通讯 2：通过提供消息传递和消息的排队机制，它可以在分布式系统环境下扩展进程间的通讯。

1.消息中间件的应用场景

跨系统数据传递高并发的流量削峰(将某些串行操作，改为并行操作)数据的分发和异步处理大数据分析与传递分布式事务

比如有一个数据要进行迁移或者请求并发过多的时候（10W的并发请求下订单），可以在这些订单入库之前，把订单请求堆积到消息队列中，让它稳健可靠的入库和执行。

2.常见的消息中间件

ActiveMQ、RabbitMQ、Kafka、RocketMQ等

3.消息中间件的本质及设计

它是一种 接收数据、接受请求，存储数据，发送数据 等功能的技术服务。 MQ消息队列：负责数据的接收、存储和传递，所以性能要优于普通服务和技术

4.消息中间件的核心组成部分

消息的协议消息的持久化机制消息的分发策略消息的高可用、高可靠消息的容错机制

四、消息队列协议

1. 什么是协议

消息中间件负责数据的传递、存储、和分发消费三个部分，数据的存储和分发的过程中肯定要遵循某种约定成俗的规范（采用底层的TCP/IP，UDP协议或其他协议）约定成俗的规范就是：协议

所谓协议是指： 1：计算机底层操作系统和应用程序通讯时共同遵守的一组约定，只有遵循共同的约定和规范，系统和底层操作系统之间才能相互交流。 2：和一般的网络应用程序不同，它主要负责数据的接受和传递，所以性能比较的高 3：计算机之间交换数据和格式，都必须严格遵守规范。

网络协议三要素：

语法：用户数据与控制信息的结构与格式，以及数据出现的顺序。语义：解释控制信息每个部分的意义。它规定了需要发出何种控制信息，以及完成的动作与做出什么样的响应。时序：时序是对事件发生顺序的详细说明。

比如我MQ发送一个信息，是以什么数据格式发送到队列中，然后每个部分的含义是什么，发送完毕以后的执行的动作，以及消费者消费消息的动作，消费完毕的响应结果和反馈是什么，然后按照对应的执行顺序进行处理。

以http请求协议为例： 1：语法：http规定了请求报文和响应报文的格式。 2：语义：客户端主动发起请求称之为请求。（这是一种定义，同时你发起的是post/get请求） 3：时序：一个请求对应一个响应。（一定先有请求在有响应，这个是时序）

消息中间件并不采用http协议。常见的消息中间件协议有：OpenWire、AMQP、MQTT、Kafka，OpenMessage协议。

面试题：为什么消息中间件不直接使用http协议呢？ 1.因为http请求报文头和响应报文头是比较复杂的，包含了cookie、数据的加密解密、状态码、响应码等附加的功能。但是对于一个消息而言，并不需要这么复杂，也没这个必要，它其实就是负责数据传递、存储、分发就行。首先追求的是高性能，尽量简洁，快速。 2.大部分情况下http都是短链接。在实际的交互过程中，一个请求到响应很有可能会中断，中断以后就不会进行持久化，造成请求的丢失。这样就不利于消息中间件的业务场景，因为消息中间件可能是一个长期的获取消息的过程，出现问题和故障要对数据或消息进行持久化等，目的是为了保证消息和数据的高可靠。

2. AMQP协议

AMQP：(全称：Advanced Message Queuing Protocol) 是高级消息队列协议。由摩根大通集团联合其他公司共同设计。是一个应用层高级消息队列协议，提供统一消息服务，为消息中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。Erlang中的实现有RabbitMQ等。 特性： 1：分布式事务支持。 2：消息的持久化支持。 3：高性能和高可靠的消息处理优势。

AMQP协议的支持者：

3. MQTT协议

MQTT协议：（Message Queueing Telemetry Transport）消息队列是IBM开放的一个即时通讯协议，物联网系统架构中的重要组成部分。 特点： 轻量、结构简单、传输快，不支持事务、没有持久化设计。 应用场景： 适用于计算能力有限、低带宽、网络不稳定的场景。

4. OpenMessage协议

是近几年由阿里、雅虎和滴滴出行、Stremalio等公司共同参与创立的分布式消息中间件、流处理等领域的应用开发标准。 特点： 结构简单、解析速度快、支持事务和持久化设计

5. Kafka协议

Kafka协议是基于 TCP/IP 的二进制协议。消息内部是通过长度来分割，由一些基本数据类型组成。 特点是： 结构简单、解析速度快、无事务支持、有持久化设计

五、消息队列持久化

常见的持久化方式：

持久化消息的保存过程： 非持久化消息： 非持久消息：是指当内存不够用的时候，会把消息和数据转移到磁盘，但是重启以后非持久化队列消息就丢失。 持久化分类： 1：队列持久化 2：消息持久化 3：交换机持久化

不论是持久化的消息还是非持久化的消息都可以写入到磁盘中，只不过非持久的是等内存不足的情况下才会被写入到磁盘中

持久化队列

代码创建持久化队列

// 参数1：名字

// 参数2：是否持久化[即durable=true。持久化的队列在web控制台中有一个D 的标记]

// 参数3：独占的queue，

// 参数4：不使用时是否自动删除，

// 参数5：其他参数

channel.queueDeclare(queueName,true,false,false,null);

web端创建持久化队列 【注】重启rabbit-server服务，持久化队列依然在，而非持久队列会丢失！

消息持久化 通过属性 deliveryMode 来设置消息是否持久化，在发送消息时通过basicPublish的参数传入：

// 参数1：交换机的名字

// 参数2：队列或者路由key

// 参数3：是否进行消息持久化

// 参数4：发送消息的内容

channel.basicPublish(exchangeName, routingKey1, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());

交换机持久化 和队列一样，交换机也需要在定义的时候设置持久化，否则在rabbit-server服务重启以后将丢失

// 参数1：交换机的名字

// 参数2：交换机的类型，topic/direct/fanout/headers

// 参数3：是否持久化

channel.exchangeDeclare(exchangeName,exchangeType,true);

六、消息的分发策略

MQ消息队列的角色：生产者、存储消息、消费者

生产者生成消息，MQ进行存储，消费者是如何获取消息的呢？一般获取数据的方式无外乎 推（push）或者 拉（pull）两种方式。典型的git就有推拉机制，http请求就是一种典型的拉取数据库数据返回的过程。消息队列MQ是一种推送的过程，而这些推机制会适用到很多的业务场景，也有很多对应推机制策略。

消息分发策略的机制和对比

ActiveMQRabbitMQKafkaRocketMQ发布订阅支持支持支持支持轮询分发支持支持支持/公平分发/支持支持/重发支持支持/支持消息拉取/支持支持支持

七、高可用&高可靠

1. 高可用

高可用：是指产品在规定的条件和时间内，处于可执行状态的能力 当业务量增加时，请求也过大，一台消息中间件服务器的会触及硬件（CPU、内存、磁盘）的极限，无法满足业务的需求。所以消息中间件必须支持集群部署，达到高可用的目的。

集群模式1 ---- Master-slave主从共享数据

【解说】：生产者讲消息发送到Master节点，所有的客户端都连接这个消息队列共享这块数据区域。Master节点负责写入，一旦Master挂掉，slave节点继续服务。从而形成高可用。

集群模式2 ---- Master- slave主从同步部署方式

【解说】写入消息同样在Master主节点上，但是主节点会同步数据到slave节点形成副本，和zookeeper或者redis主从机制很类同。这样可以达到负载均衡的效果，如果消费者有多个，就可以去不同的节点消费，因为消息的拷贝和同步会占用很大的带宽和网络资源。在后续的 RabbitMQ 中会有使用

集群模式3 ---- Master- slave主从同步部署方式

【解说】和上面的区别不大，但可以在任意节点进行写入

集群模式4 ---- Master- slave主从同步部署方式

【解说】如果你插入的数据是broker-1中，元数据信息会存储数据的相关描述和记录存放的位置（队列）。它会对描述信息（元数据信息）进行同步，如果消费者在broker-2中进行消费，几乎没有对应的消息。可以从对应的元数据信息中去查询，然后返回对应的消息信息。 场景：比如买火车票或者黄牛买演唱会门票，比如第一个黄牛有顾客说要买的演唱会门票，但是没有但是他会去联系其他的黄牛询问，如果有就返回。

集群模式5 ---- Master-slave与Breoker-cluster组合的方案

【解说】实现多主多从的模式，来完成消息的高可用以及数据的热备机制。在生产规模达到一定的阶段的时候，这种使用的频率比较高。

集群模式最终目的都是为保证：消息服务器不会挂掉，出现了故障依然可以保证消息服务继续使用。

终归三点： 要么消息共享；要么消息同步；要么元数据共享

2. 高可靠

高可用：指系统可以无故障低持续运行。比如一个系统突然崩溃，报错，异常等等并不影响线上业务的正常运行，出错的几率极低，就称之为：高可靠。 在高并发的业务场景中，如果不能保证系统的高可靠，那造成的隐患和损失是非常严重的。 如何保证中间件消息的可靠性呢？可以从两个方面考虑： 1：消息的传输：通过协议来保证系统间数据解析的正确性。 2：消息的存储可靠：通过持久化来保证消息的可靠性。

八、 RabbitMQ入门及安装

RabbitMQ官网：https://www.rabbitmq.com/

RabbitMQ is the most widely deployed open source message broker. With tens of thousands of users, RabbitMQ is one of the most popular open source message brokers. From T-Mobile to Runtastic, RabbitMQ is used worldwide at small startups and large enterprises. RabbitMQ is lightweight and easy to deploy on premises and in the cloud. It supports multiple messaging protocols. RabbitMQ can be deployed in distributed and federated configurations to meet high-scale, high-availability requirements. RabbitMQ runs on many operating systems and cloud environments, and provides a wide range of developer tools for most popular languages.RabbitMQ是部署最广泛的开源消息代理。 RabbitMQ拥有数以万计的用户，是最受欢迎的开源消息代理之一。从T-Mobile到Runtastic，RabbitMQ在全球范围内用于小型初创企业和大型企业。 RabbitMQ是轻量级的，易于在本地和云中部署。它支持多种消息传递协议。RabbitMQ可以部署在分布式和联合配置中，以满足高规模、高可用性的要求。 RabbitMQ在许多操作系统和云环境上运行，并为大多数流行语言提供了广泛的开发工具。

【概述】：RabbitMQ是一个开源的、遵循AMQP协议的、基于Erlang语言编写的消息代理，支持多种客户端（语言）。用于在分布式系统中 存储、转发消息，具有高可用、高可扩性、易用性等特征。

1. 安装RabbitMQ

官方安装包下载地址 rabbitmq下载地址：https://github.com/rabbitmq/rabbitmq-server/releases erlang下载地址：hhttps://hub.fastgit.org/rabbitmq/erlang-rpm/releases socket下载地址：http://www.rpmfind.net/linux/rpm2html/search.php?query=socat(x86-64)

1：下载地址：https://www.rabbitmq.com/download.html 2：环境准备：CentOS7.x+ / Erlang RabbitMQ是采用Erlang语言开发的。所以系统必须提供Erlang环境，先安装 Erlang

Erlang与RabbitMQ的版本对应：https://www.rabbitmq.com/which-erlang.html

安装 Erlang

下载地址：https://www.erlang-solutions.com/downloads/ 阿里云centos7.x中下载erlang。注：一定是 el7版本： 网页地址：https://packagecloud.io/rabbitmq/erlang

wget --content-disposition "https://packagecloud.io/rabbitmq/erlang/packages/el/7/erlang-23.3.4.11-1.el7.x86_64.rpm/download.rpm?distro_version_id=140"

安装

rpm -ivh erlang-23.3.4.11-1.el7.x86_64.rpm

测试 erl -v

安装 socat

rpm -ivh socat-1.7.3.2-2.el7.x86_64.rpm

【注意】centos7系统，要下载 el7 版本： socket下载地址：http://www.rpmfind.net/linux/rpm2html/search.php?query=socat(x86-64)

安装 RabbitMQ

下载网站：https://packagecloud.io/app/rabbitmq/rabbitmq-server

使用wget的方式，获取安装包

wget --content-disposition "https://packagecloud.io/rabbitmq/rabbitmq-server/packages/el/7/rabbitmq-server-3.10.0-1.el7.noarch.rpm/download.rpm?distro_version_id=140"

安装

rpm -ivh rabbitmq-server-3.10.0-1.el7.noarch.rpm

启动、关停

# 启动服务>

systemctl start rabbitmq-server

# 查看服务状态>

systemctl status rabbitmq-server

# 停止服务>

systemctl stop rabbitmq-server

# 开机启动服务>

systemctl enable rabbitmq-server

端口说明

端口说明5672RabbitMQ的通讯端口25672RabbitMQ的节点间的CLI通讯端口15672RabbitMQ HTTP_API的端口，管理员用户才能访问。用于管理RabbitMQ，需要启动Management插件。1883、8883MQTT插件启动时的端口61613、61614STOMP客户端插件启用的时候的端口15674、15675基于webscoket的STOMP端口和MOTT端口

2. Web管理界面及授权

管理界面

默认情况下，rabbitmq没有安装web端插件，需要安装才可以生效

rabbitmq-plugins enable rabbitmq_management

安装完毕，重启服务

systemctl restart rabbitmq-server

开启防火墙，通过公网访问15672端口，得到页面如下：

授权用户

rabbitmq有一个默认账号和密码是：guest 。默认情况只能在localhost本机下访问，所以需要添加一个远程登录的用户：

新增用户、密码

rabbitmqctl add_user admin admin

设置权限

rabbitmqctl set_user_tags admin administrator

用户级别及权限：

角色权限none不能访问management pluginmanagement查看自己相关节点信息 列出自己可以通过AMQP登入的虚拟机 查看自己的虚拟机节点 virtual hosts的queues、exchanges和bindings信息 查看和关闭自己的channels和connections 查看有关自己的虚拟机节点virtual hosts的统计信息（包括其他用户在这个节点中的活动信息）Policymaker包含management所有权限查看和创建和删除自己的virtual hosts所属的policies和parameters信息Monitoring包含management所有权限罗列出所有的virtual hosts，包括不能登录的virtual hosts。查看其他用户的connections和channels信息查看节点级别的数据如clustering和memory使用情况查看所有的virtual hosts的全局统计信息Administrator可以创建和删除virtual hosts可以查看，创建和删除users查看创建permisssions关闭所有用户的connections

为用户添加资源权限

rabbitmqctl set_permissions -p / admin ".*" ".*" ".*"

使用admin登录web端页面： 命令小结

rabbitmqctl add_user 账号 密码 # 添加用户

rabbitmqctl set_user_tags 账号 administrator # 授权用户

rabbitmqctl change_password Username Newpassword # 修改密码

rabbitmqctl delete_user Username # 删除用户

rabbitmqctl list_users # 查看用户清单

# 为用户设置administrator角色

rabbitmqctl set_permissions -p / 用户名 ".*" ".*" ".*"

3. Docker安装RabbitMQ

参考网站： 1：https://www.rabbitmq.com/download.html 2：https://registry.hub.docker.com/_/rabbitmq/

【注】rabbitmq:management 版本默认支持web端插件；rabbitmq:latest 版本提交最多，但需手动开启网页插件

# 1. 拉取镜像

docker pull rabbitmq:management

# 2. 创建并运行容器

# —hostname：指定容器主机名称

# —name:指定容器名称

# -p:将mq端口号映射到主机

# -e RABBITMQ_DEFAULT_USER= 设置用户名

# -e RABBITMQ_DEFAULT_PASS= 设置密码

docker run -d --name myrabbit -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin -p 15672:15672 -p 5672:5672 -p 25672:25672 -p 61613:61613 -p 1883:1883 rabbitmq

# 3. 开启网页客户端

docker exec 容器id rabbitmq-plugins enable rabbitmq_management`

九、原理讲解

1. AMQP协议

AMQP：（Advanced Message Queuing Protocol）高级消息队列协议。是应用层协议的一个开发标准，为面向消息的中间件设计

生产者与消费者流转过程如下：

为什么RabbitMQ是基于channel而不是连接的操作?

无论是生产者还是消费者，都需要和 RabbitMQ Broker 建立连接，这个连接就是一条 TCP 连接，也就是 Connection。一旦 TCP 连接建立起来，客户端紧接着可以创建一个 AMQP 信道（Channel），每个信道都会被指派一个唯一的 ID。Channel 是建立在 Connection 之上的虚拟连接，RabbitMQ 处理的每条 AMQP 指令都是通过信道完成的。

我们完全可以使用 Connection 就能完成信道的工作，为什么还要引入信道呢？

试想这样一个场景，一个应用程序中有很多个线程需要从 RabbitMQ 中消费消息，或者生产消息，那么必然需要建立很多个 Connection，也就是多个 TCP 连接； 然而对于操作系统而言，建立和销毁 TCP 连接是非常昂贵的开销，如果遇到使用高峰，性能瓶颈也随之显现。 RabbitMQ 采用类似 NIO（Non-blocking I/O）的做法，选择 TCP 连接复用，不仅可以减少性能开销，同时也便于管理； 每个线程把持一个信道，所有信道复用 Connection 的 TCP 连接。同时 RabbitMQ 可以确保每个线程的私密性，就像拥有独立的连接一样。 当每个信道的流量不是很大时，复用单一的 Connection 可以在产生性能瓶颈的情况下有效地节省 TCP 连接资源；但是 信道本身的流量很大时，这时候多个信道复用一个 Connection 就会产生性能瓶颈，进而使整体的流量被限制了。此时就需要开辟多个 Connection，将这些信道均摊到这些 Connection 中，至于这些相关的调优策略需要根据业务自身的实际情况进行调节。 信道在 AMQP 中是一个很重要的概念，大多数操作都是在信道这个层面展开的： 比如 channel.exchangeDeclare、channel.queueDeclare、channel.basicPublish、channel.basicConsume 等方法； RabbitMQ 相关的 API 与 AMQP 紧密相连，比如 channel.basicPublish 对应 AMQP 的 Basic.Publish 命令。

2. 核心组成部分

核心概念：

Server：又称Broker。接受客户端的连接，实现AMQP实体服务。 安装rabbitmq-server Connection：连接，应用程序与Broker的网络连接 。TCP/IP 三次握手和四次挥手 Channel：网络信道，几乎所有的操作都在Channel中进行，Channel是进行消息读写的通道，客户端可以建立对各Channel，每个Channel代表一个会话任务。 Message：消息：服务与应用程序之间传送的数据，由Properties和body组成。Properties可是对消息进行修饰，比如消息的优先级，延迟等高级特性；Body则就是消息体的内容。 Virtual Host：虚拟地址，用于进行逻辑隔离，最上层的消息路由，一个虚拟主机理由可以有若干个Exchange和Queue，同一个虚拟主机里面不能有相同名字的Exchange Exchange：交换机，接受消息，根据路由键发送消息到绑定的队列。(不具备消息存储的能力) Bindings：Exchange和Queue之间的虚拟连接，binding中可以保护多个routing key.交换机和队列之间的绑定。 Routing key：是一个路由规则，虚拟机可以用它来确定如何路由一个特定消息。消费者可能有多个，队列中的消息如果想要指定推送呢？就要通过这个key来查找投递给谁。 Queue：队列，也称为消息队列（Message Queue），保存消息并将它们转发给消费者

整体架构

运行流程

十、 RabbitMQ模式

1. 简单模式

创建maven项目 导入依赖

com.rabbitmq

amqp-client

5.10.0

org.springframework.amqp

spring-amqp

2.2.5.RELEASE

org.springframework.amqp

spring-rabbit

2.2.5.RELEASE

org.springframework.boot

spring-boot-starter-amqp

2.3.0.RELEASE

编写生产者

public class Producer {

public static void main(String[] args) {

//所有的中间价技术都是基于tcp/ip协议的，并在此协议上构建的。rabbitmq遵循的是amqp协议。

//所以既然如此，必然会有ip和port

//1.创建连接工程

ConnectionFactory connectionFactory = new ConnectionFactory();

connectionFactory.setHost("http://39.104.144.107");

connectionFactory.setPort(5672);

connectionFactory.setUsername("xinan");

connectionFactory.setPassword("xinan2");

connectionFactory.setVirtualHost("/");

Connection connection = null;

Channel channel = null;

try{

//2.创建Connection,名字叫做Producer

connection = connectionFactory.newConnection("Producer");

//3.通过连接获取通道channel

channel = connection.createChannel();

//4.声明队列

String queueName = "queue1";

//声明队列时的五个参数

/**

* @params1:队列名称

* @params2:是否要持久化 durable=false,所谓持久化消息就是 是否会存盘，

* false：非持久化，true:持久化，非持久化会存盘吗？

* @params3:排他性，是否是一个独占队列

* @params4:最后一个消费者消费完消息之后，是否自动把队列删除

* @params5:携带一些附加的参数

*/

channel.queueDeclare(queueName,false,false,false,null);

//5.准备消息内容

String message = "Hello world";

//6.发送消息给队列

channel.basicPublish("",queueName,null,message.getBytes(StandardCharsets.UTF_8));

System.out.println("消息发送成功");

}

catch (Exception e){

e.printStackTrace();

}

finally {

try {

if (channel != null && channel.isOpen()) {

channel.close(); //7.关闭通道

}

if(connection != null && connection.isOpen()) {

connection.close(); //8.关闭连接

}

}

catch (Exception e){

e.printStackTrace();

}

}

}

}

执行上面代码，可以在web控制台看到 queue1 队列的信息 编写消费者代码

public class Consumer {

public static void main(String[] args) {

//1.创建连接工程

ConnectionFactory connectionFactory = new ConnectionFactory();

connectionFactory.setHost("39.104.144.107");

connectionFactory.setPort(5672);

connectionFactory.setUsername("admin");

connectionFactory.setPassword("admin");

connectionFactory.setVirtualHost("/");

Connection connection = null;

Channel channel = null;

try{

//2.创建Connection,名字叫做Producer

connection = connectionFactory.newConnection("Producer");

//3.通过连接获取通道channel

channel = connection.createChannel();

//4.创建消费者

channel.basicConsume("queue1", true, new DeliverCallback() {

@Override

public void handle(String consumerTag, Delivery message) throws IOException {

System.out.println("收到的消息是： " + new String(message.getBody(), "UTF-8"));

}

}, new CancelCallback() {

@Override

public void handle(String s) throws IOException {

System.out.println("接收消息失败。。。");

}

});

System.out.println("开始接收消息");

//在这里暂停一下，相当于getchar()

System.in.read();

}

catch (Exception e){

e.printStackTrace();

}

finally {

try {

if (channel != null && channel.isOpen()) {

channel.close(); //7.关闭通道

}

if(connection != null && connection.isOpen()) {

connection.close(); //8.关闭连接

}

}

catch (Exception e){

e.printStackTrace();

}

}

}

}

测试结果

持久化队列和非持久化队列的区别

持久化队列：在rabbitMQ服务重启后，该队列仍然存在，队列中的消息也会存在非持久化队列：队列只存在于rabbitMQ的运行过程中，服务重启后，队列不会存在；队列中的消息在rabbitMQ的运行过程中会持久化，但是在rabbitMQ服务重启后，持久化的消息也会消失。

2. 四种 exchage

1）fanout类型（发布与订阅模式）

交换机为 fanout 类型，通过交换机，向他下面的队列都发送一样的消息；指定路由key是没有意义的，仍然会所有的队列都会收到消息。

通过图像化界面操作

创建交换机 创建队列 交换机与队列绑定 发布消息 Ready、Total 数量 +1

核心代码

消费者1：

// 1. 声明交换机

String exchangeName = "fanout_exchang"; // 交换机名称

String type = "fanout"; // 类型

channel.exchangeDeclare(exchangeName, type);

// 2. 声明队列：订阅者fanout_queue1

String queueName = "fanout_queue1";

// 队列名，持久化，排他性，自动删除，携带额外的参数

channel.queueDeclare(queueName, false, false, false, null);

// 3. 队列&交换机绑定：其中 routingkey(也称之为 binding key)为空字符串

channel.queueBind(queueName, exchangeName, "");

// 4. 队列消费消息

channel.basicConsume(queueName, true, DeliverCallback(),CancelCallback());

消费者2：

// 1. 声明交换机

String exchangeName = "fanout_exchang"; // 交换机名称

String type = "fanout"; // 类型

channel.exchangeDeclare(exchangeName, type);

// 2. 声明队列：订阅者fanout_queue2

String queueName = "fanout_queue2";

// 队列名，持久化，排他性，自动删除，携带额外的参数

channel.queueDeclare(queueName, false, false, false, null);

// 3. 队列&交换机绑定：其中 routingkey(也称之为 binding key)为空字符串

channel.queueBind(queueName, exchangeName, "");

// 4. 队列消费消息

channel.basicConsume(queueName, true, DeliverCallback(),CancelCallback());

生产者：

// 1. 声明交换机：名称与类型

String exchangeName = "fanout_exchang"; //交换机名称

String exchangeType = "fanout"; //交换机类型

channel.exchangeDeclare(exchangeName, exchangeType); //声明交换机

// 向fanout_exchange交换机，发布消息

String message = "hello world , fanout!!!";

channel.basicPublish("fanout_exchang", "", null, message .getBytes(StandardCharsets.UTF_8));

2）Direct类型（Routing模式）

direct类型的交换机，通过队列设置不同的 Routing key ，来接受不同的消息； 交换机在发消息的时候，通过指定的不同的 Routing key ，来转发到指定的 Routing key的队列。

消费者1

//2.创建Connection,名字叫做Producer

connection = connectionFactory.newConnection("Producer");

//3.通过连接获取通道channel

channel = connection.createChannel();

//4.声明队列

String queueName = "routing_queue1";

channel.queueDeclare(queueName, false, false, false, null);

//5.绑定交换机与队列

String exchangeName = "direct_exchange";

String routingkey = "error";

channel.queueBind(queueName, exchangeName, routingkey);

//6.创建消费者

channel.basicConsume(queueName, true, DeliverCallback(),CancelCallback());

消费者2

//2.创建Connection,名字叫做Producer

connection = connectionFactory.newConnection("Producer");

//3.通过连接获取通道channel

channel = connection.createChannel();

//4.声明队列

String queueName = "routing_queue2";

channel.queueDeclare(queueName, false, false, false, null);

//5.绑定交换机与队列

String exchangeName = "direct_exchange";

String routingkey = "info";

channel.queueBind(queueName, exchangeName, routingkey);

//6.消费消息

channel.basicConsume(queueName, true, DeliverCallback(),CancelCallback());

生产者

//2.创建Connection,名字叫做Producer

connection = connectionFactory.newConnection("Producer");

//3.通过连接获取通道channel

channel = connection.createChannel();

//4.声明交换机（如web端已操作，则省略）

String exchangeName = "direct_exchange";

String type = "direct";

channel.exchangeDeclare(exchangeName, type);

//5.准备消息内容

String message = "Hello world!!!! direct";

String routingKey = "info";

//6.发送消息给队列

channel.basicPublish(exchangeName,routingKey ,null,message.getBytes(StandardCharsets.UTF_8));

message.getBytes();

3）topic类型（主题模式）

设置交换机为topic类型，通过 Routing key 模式匹配来分发队列里面的消息

* : 匹配一个单词# : 匹配 0个或多个单词

表达式含义*.orange.*前面有1级，orange，后面有1级*.*.rabbit前面有1级，中间有1级，rabbitlazy.#lazy后面可以有0个或多个，都是匹配的

生产者

//2.创建Connection,名字叫做Producer

connection = connectionFactory.newConnection("Producer");

//3.通过连接获取通道channel

channel = connection.createChannel();

//4.声明交换机（如web端已操作，则省略）

String exchangeName = "topic_exchange";

String type = "topic";

channel.exchangeDeclare(exchangeName, type);

//5.准备消息内容

String message = "Hello world!!!! topic1"; //发送的消息

String routingkey = "com.orange.rabbit"; //routingKey

String routingkey2 = "lazy.xinan"; //routingKey

//6.发送消息给队列

channel.basicPublish(exchangeName,routingkey,null,message.getBytes(StandardCharsets.UTF_8));

消费者1

//2.创建Connection,名字叫做Producer

connection = connectionFactory.newConnection("Producer");

//3.通过连接获取通道channel

channel = connection.createChannel();

//4.声明队列

String queueName = "topic_queue1";

channel.queueDeclare(queueName, false, false, false, null);

//5.绑定交换机与队列

String exchangeName = "topic_exchange";

String routingkey = "*.orange.*";

channel.queueBind(queueName, exchangeName, routingkey);

//6.消费消息

。。。。。。

消费者2

//2.创建Connection,名字叫做Producer

connection = connectionFactory.newConnection("Producer");

//3.通过连接获取通道channel

channel = connection.createChannel();

//4.声明队列

String queueName = "topic_queue2";

channel.queueDeclare(queueName, false, false, false, null);

//5.绑定交换机与队列

String exchangeName = "topic_exchange";

String routingkey1 = "*.*.rabbit";

String routingkey2 = "lazy.#";

channel.queueBind(queueName, exchangeName, routingkey1);

channel.queueBind(queueName, exchangeName, routingkey2);

//6.消费消息

。。。。。。

4）Headers类型 【几乎不用】

交换机为Headers模式，队列设置不同的参数； 发送消息时，通过不同的参数，将消息转达到不同的队列

队列Arguments举例结果queue1x:1 y:1x=1queue2能收到queue2x:1x=1，y=1queue1、queue2能收到queue3x:2 y:1x=2都收不到

生产者

//2.创建Connection,名字叫做Producer

connection = connectionFactory.newConnection("Producer");

//3.通过连接获取通道channel

channel = connection.createChannel();

//4.声明交换机（如web端已操作，则省略）

String exchangeName = "topic_exchange_header";

String type = "headers";

channel.exchangeDeclare(exchangeName, type);

//5.准备消息内容

String message = "Hello world!!!! topic_header8-----------"; //发送的消息

//6.定义 header

Map header = new HashMap();

header.put("x","1");

header.put("y","1");

AMQP.BasicProperties.Builder properties = new AMQP.BasicProperties().builder().headers(header);

//7.发送消息给队列

channel.basicPublish(exchangeName,"",properties.build(),message.getBytes(StandardCharsets.UTF_8));

消费者1

//4.声明队列

String queueName = "topic_header_queue1";

channel.queueDeclare(queueName, false, false, false, null);

//5.定义 header

Map header = new HashMap();

//x-match: all表所有key-value全部匹配才匹配成功 ，any表只需要匹配任意一个key-value 即匹配成功。

header.put("x-match", "all");

header.put("x","1");

header.put("y","2");

//6.绑定交换机与队列

String exchangeName = "topic_exchange_header";

channel.queueBind(queueName, exchangeName,"", header); //绑定的时候，传入 header

//7.创建消费者消费消息

DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {

@Override

public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

String message = new String(body, "UTF-8");

System.out.println(message);

}

};

//8.消费消息

channel.basicConsume(queueName, true, defaultConsumer);

消费者2

//4.声明队列

String queueName = "topic_header_queue2";

channel.queueDeclare(queueName, false, false, false, null);

//5.定义 header

Map header = new HashMap();

//x-match: all表所有key-value全部匹配才匹配成功 ，any表只需要匹配任意一个key-value 即匹配成功。

header.put("x-match", "all");

header.put("x","1");

//6.绑定交换机与队列

String exchangeName = "topic_exchange_header";

channel.queueBind(queueName, exchangeName, "",header);

//7.创建消费者消费消息

DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {

@Override

public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

String message = new String(body, "UTF-8");

System.out.println(message);

}

};

//8.消费消息

channel.basicConsume(queueName, true, defaultConsumer);

3. Work模式

当有多个消费者时，消息会被哪个消费者消费呢？该如何均衡消费者消费信息的多少呢? 主要有两种模式： 1、轮询分发：一个消费者一条，按均分配； 2、公平分发：根据消费者的消费能力进行公平分发。处理快的处理的多，处理慢的处理的少；

1）轮询模式

特点：当有多个消费者接入时，消息的分配模式是一个消费者分配一条，直至消息消费完成；并不会因为某个服务器的效率低，而出现不均等分发的现象。

生产者

生产者使用默认交换机，发布消息时，填写第二个参数【指明发动到哪个队列】生产者使用自定义交换机，发布消息时，不要填写第二个参数【此时代表routingKey】

//2.创建Connection,名字叫做Producer

connection = connectionFactory.newConnection("Producer");

//3.通过连接获取通道channel

channel = connection.createChannel();

//4.声明交换机

String exchangeName = "work_exchange_polling";

String type = "direct";

channel.exchangeDeclare(exchangeName, type);

//5.声明队列

String queueName = "work_polling_queue";

channel.queueDeclare(queueName, false, false, false, null);

//6.绑定队列与交换机

channel.queueBind(queueName,exchangeName,"");

for (int i = 1; i <= 20; i++) {

//消息的内容

String msg = "hello: " + i;

// 7: 发送消息给中间件rabbitmq-server

// @params1: 交换机 exchange

// @params2: 队列名称 / routingkey

// 如果指定了交换机名称，第二个参数代表 routingKey;

// 如果没指定交换机名称，第二个参数代表 队列名称

// @params3: 属性配置

// @params4: 发送消息的内容

channel.basicPublish(exchangeName, "", null, msg.getBytes());

}

消费者

消费者1、2代码相同，消费的是同一个队列【queueName相同】channel.basicQos(1);，指定每次轮询，消费多少条消息

//4.要消费的队列

String queueName = "work_polling_queue";

//5.绑定交换机与队列

String exchangeName = "work_exchange_polling";

channel.queueBind(queueName, exchangeName,"");

//6.创建消费者消费消息

DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {

@Override

public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

String message = new String(body, "UTF-8");

System.out.println(message);

}

};

//7.每次分发多少条消息

channel.basicQos(1);

//8.消费消息(轮询分发，@param2 true代表自动应答)

channel.basicConsume(queueName, true, new DeliverCallback() {

@Override

public void handle(String consumerTag, Delivery message) throws IOException {

System.out.println("收到的消息是： " + new String(message.getBody(), "UTF-8"));

}

}, new CancelCallback() {

@Override

public void handle(String s) throws IOException {

System.out.println("接收消息失败。。。");

}

});

//【模拟服务器延迟】消费者1 沉睡1000ms ; 消费者2 沉睡200ms

Thread.sleep(1000);

2）公平模式

生产者代码，与轮询分发完全一样消费者 （1） 消费者1、2代码相同，消费的是同一个队列【queueName相同】 （2） 采用手动应答的方式，每次消费一条消息

//2.创建Connection,名字叫做Producer

connection = connectionFactory.newConnection("Producer");

//3.通过连接获取通道channel

channel = connection.createChannel();

//4.要消费的队列

String queueName = "work_polling_queue";

//5.绑定交换机与队列

String exchangeName = "work_exchange_polling";

channel.queueBind(queueName, exchangeName,"");

//6.创建消费者消费消息

DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {

@Override

public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

String message = new String(body, "UTF-8");

System.out.println(message);

}

};

//7.定义接收消息的回调

Channel finalChannel = channel;

finalChannel.basicQos(1); //每次消费一条

//8.消费消息（第二个参数false: 采取手动应答）

channel.basicConsume(queueName, false, new DeliverCallback() {

@Override

public void handle(String consumerTag, Delivery message) throws IOException {

try {

System.out.println("收到的消息是： " + new String(message.getBody(), "UTF-8"));

Thread.sleep(1000); // 两个消费者，一个沉睡1000ms，一个沉睡 200ms

// 两个参数：应答方式、是否批量消费

finalChannel.basicAck(message.getEnvelope().getDeliveryTag(),false);

} catch (Exception e) {

e.printStackTrace();

}

}

}, new CancelCallback() {

@Override

public void handle(String s) throws IOException {

System.out.println("接收消息失败。。。");

}

});

十一、使用场景

解耦、削峰、异步

1、串行方式：将订单信息写入数据库成功后，发送注册邮件，再发送注册短信。以上三个任务全部完成后，返回给客户端。如果串行执行，只有三个任务全部完成，业务才能完成。

/**

* 串行执行代码：

* 如果有事务，任何一个发送任务出错，都可能造成事务的回滚

*/

public void makeOrder(){

// 1 :保存订单

orderService.saveOrder();

// 2： 发送短信服务

messageService.sendSMS("order");//1-2 s

// 3： 发送email服务

emailService.sendEmail("order");//1-2 s

// 4： 发送APP服务

appService.sendApp("order");

}

2、并行方式：将订单信息写入数据库成功后，发送注册邮件的同时，发送注册短信。以上三个任务完成后，返回给客户端。与串行的差别是，并行的方式可以提高处理的时间

/**

* 异步执行代码：

* 采用线程池的方式，同时发送三种信息

*/

public void makeOrder(){

// 1 :保存订单

orderService.saveOrder();

// 相关发送

relationMessage();

}

public void relationMessage(){

// 1：异步发送短信服务

theadpool.submit(new Callable