新余网站建设人员,自建站怎么做,中国数学外国人做视频网站,新闻门户网站免费建设目录
一、RabbitMQ-01
1.1 MQ概述
1.2 MQ的优势和劣势
1.2.1 优势
1.2.2 劣势
1.2.3 MQ应用场景
1.2.4 常用的MQ产品
1.3 RabbitMQ的基本介绍
1.3.1 AMQP介绍
1.3.2 RabbitMQ基础架构
1.3.3 RabbitMQ的6种工作模式
编辑
1.4 AMQP和JMS
1.4.1 AMQP
1.4.2 JMS
…目录
一、RabbitMQ-01
1.1 MQ概述
1.2 MQ的优势和劣势
1.2.1 优势
1.2.2 劣势
1.2.3 MQ应用场景
1.2.4 常用的MQ产品
1.3 RabbitMQ的基本介绍
1.3.1 AMQP介绍
1.3.2 RabbitMQ基础架构
1.3.3 RabbitMQ的6种工作模式
编辑
1.4 AMQP和JMS
1.4.1 AMQP
1.4.2 JMS
1.4.3 AMQP与 JMS 区别
1.4.4 市场常见消息队列
1.5 RabbitMQ的安装和配置
1.6 使用图形化添加一个用户和虚拟主机
1.6.1 添加一个用户
1.6.2 添加一个虚拟主机
1.6.3 配置用户和虚拟主机
1.6.4 删除一个虚拟主机
1.6.5 使用可视化完成消息收发
1.7 java测试RabbitMQ的工作模式
1.7.1 简单模式(Hello world)
1.7.2 工作队列模式(Work queues)
1.7.3 发布订阅模式(Publish/Subscribe)
1.7.4 路由模式(Routing)
1.7.5 通配符模式(Topics)
1.7.6 模式间的异同
1.7.7 使用可视化实现主题模式
1.8 SpringBoot项目整合RabbitMQ
1.8.1 介绍
1.8.2 创建父工程
1.8.3 搭建生产者工程
1.8.4 搭建消费者工程 一、RabbitMQ-01
1.1 MQ概述
MQ全称 Message Queue[kjuː]消息队列是在消息的传输过程中保存消息的容器。多用于分布式系统之间进行通信。队列是一种容器用于存放数据的都是容器存放消息的就是消息队列
分布式系统的调用有两种方式
1直接调用
B系统直接调用A系统
被调用者叫生产者 调用者是消费者 2间接调用
A将数据存放到中间一个系统通过中间的系统发送到B
MQ是用于存放消息的中间件 1.2 MQ的优势和劣势
1.2.1 优势
应用解耦提高系统容错性和可维护性。
异步提速提升用户体验和系统吞吐量。
削峰填谷提高系统稳定性。
1应用解耦
使用前系统的耦合性越高容错性就越低可维护性就越低。
例访问 订单系统 的时候依赖于库存系统、支付系统、物流系统①当库存系统发生异常就有可能导致订单系统发生异常下单失败②追加系统(X)就只能修改订单系统更改代码导致维护性比较低 使用后使用 MQ 使得应用间解耦提升容错性和可维护性
①库存系统宕机订单系统影响不大因为消息已经发送到mq了当库存系统恢复的时候就可以正常使用了。
②追加系统的时候跟订单系统无关 2异步提速
使用前
一个下单操作耗时20 300 300 300 920ms
用户点击完下单按钮后需要等待920ms才能得到下单响应太慢 使用后
用户点击完下单按钮后只需等待25ms就能得到下单响应 (20 5 25ms)。
提升用户体验和系统吞吐量单位时间内处理请求的数目。
以前920ms处理一个请求现在25ms处理一个请求系统的吞吐量增加 3削峰填谷(一般是削峰)
使用前
一次性来5000个请求会导致系统崩溃 使用后
使用了 MQ 之后限制消费消息的速度为1000这样一来高峰期产生的数据势必会被积压在 MQ 中高峰就被“削”掉了但是因为消息积压在高峰期过后的一段时间内消费消息的速度还是会维持在1000直到消费完积压的消息这就叫做“填谷”。
使用MQ后可以提高系统稳定性。 1.2.2 劣势 1系统可用性降低
系统引入的外部依赖越多系统稳定性越差。一旦 MQ 宕机就会对业务造成影响。如何保证MQ的高可用
2系统复杂度提高
MQ 的加入大大增加了系统的复杂度以前系统间是同步的远程调用现在是通过 MQ 进行异步调用。如何保证消息没有被重复消费怎么处理消息丢失情况那么保证消息传递的顺序性
3一致性问题
A 系统处理完业务通过 MQ 给B、C、D三个系统发消息数据如果 B 系统、C 系统处理成功D 系统处理失败。如何保证消息数据处理的一致性
1.2.3 MQ应用场景
1、生产者不需要从消费者处获得反馈。引入消息队列之前的直接调用其接口的返回值应该为空这才让明明下层的动作还没做上层却当成动作做完了继续往后走即所谓异步成为了可能。
2、容许短暂的不一致性。
3、确实是用了有效果。即解耦、提速、削峰这些方面的收益超过加入MQ管理MQ这些成本。
1.2.4 常用的MQ产品 目前业界有很多的 MQ 产品例如 RabbitMQ、RocketMQ、ActiveMQ、Kafka、ZeroMQ、MetaMq等也有直接使用 Redis 充当消息队列的案例而这些消息队列产品各有侧重在实际选型时需要结合自身需求及 MQ 产品特征综合考虑。 1.3 RabbitMQ的基本介绍
1.3.1 AMQP介绍
AMQP即 Advanced Message Queuing Protocol英[ˈprəʊtəkɒl]高级消息队列协议是一个网络协议是应用层协议的一个开放标准为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息并不受客户端/中间件不同产品不同的开发语言等条件的限制。2006年AMQP 规范发布。类比HTTP。 消息队列中间件
exchange 交换机用来分发消息到不同的容器 queue通过路由来处理
queue 容器(队列)
routes 路由
流程
生产者(procedure)发布消息到交换机(exchange)
交换机(exchange)通过不同的路由规则发布/路由 给不同的queue进行存储
消费者(consumer)通过队列去监听拿到消息进行消费
2007年Rabbit 技术公司基于 AMQP 标准开发的 RabbitMQ 1.0 发布。RabbitMQ 采用 Erlang 语言开发。Erlang 语言由 Ericson 设计专门为开发高并发和分布式系统的一种语言在电信领域使用广泛。
1.3.2 RabbitMQ基础架构
相关概念介绍(如下图)
1、Broker接收和分发消息的应用RabbitMQ Server就是 Message Broker。
2、Virtual host出于多租户和安全因素设计的把 AMQP 的基本组件划分到一个虚拟的分组中类似于网络中的 namespace 概念。当多个不同的用户使用同一个 RabbitMQ server 提供的服务时可以划分出多个vhost每个用户在自己的 vhost 创建 exchangequeue 等。
3、Connectionpublisherconsumer 和 broker 之间的 TCP 连接。
4、Channel如果每一次访问 RabbitMQ 都建立一个 Connection在消息量大的时候建立 TCP Connection的开销将是巨大的效率也较低。Channel 是在 connection 内部建立的逻辑连接如果应用程序支持多线程通常每个thread创建单独的 channel 进行通讯AMQP method 包含了channel id 帮助客户端和message broker 识别 channel所以 channel 之间是完全隔离的。Channel 作为轻量级的 Connection 极大减少了操作系统建立 TCP connection 的开销。
5、Exchangemessage 到达 broker 的第一站根据分发规则匹配查询表中的 routing key分发消息到queue 中去。
常用的类型有
direct (point-to-point)
topic (publish-subscribe)
fanout (multicast)
6、Queue消息最终被送到这里等待 consumer 取走
7、Bindingexchange 和 queue 之间的虚拟连接binding 中可以包含 routing key。Binding 信息被保存到 exchange 中的查询表中用于 message 的分发依据。
① procedure 和consumer都是客户端客户端通过链接和服务端进行通信所以需要建立起来连接然后进行通信
② 一个rabbitmq里面有很多的虚拟机 相当于mysql里面有很多数据库数据库里面有很多表都是独立的。
③ 每个虚拟机里面有很多的exchange和queue独立分区的作用 1.3.3 RabbitMQ的6种工作模式
RabbitMQ 提供了 6 种工作模式
1、Hello World (简单模式)
2、work queues (工作队列模式)
3、Publish/Subscribe (发布与订阅模式)
4、Routing (路由模式)
5、Topics (主题模式)
6、RPC (远程调用模式)远程调用不太算 MQ暂不作介绍
官网对应模式介绍RabbitMQ Tutorials — RabbitMQ 1.4 AMQP和JMS
MQ是消息通信的模型
实现MQ的大致有两种主流方式AMQP、JMS。
1.4.1 AMQP
AMQP是一种协议更准确的说是一种binary wire-level protocol链接协议。这是其和 JMS的本质差别AMQP不从API层进行限定而是直接定义网络交换的数据格式。
1.4.2 JMS
JMS即Java消息服务Java Message Service应用程序接口是一个Java平台中关于面向消息中间件MOM的API用于在两个应用程序之间或分布式系统中发送消息进行异步通信。规定了消息客户端的一套api的东西rabbitmq没有遵循JMS规则
JMS 是 JavaEE 规范中的一种类比JDBC。
1.4.3 AMQP与 JMS 区别
JMS是定义了统一的接口来对消息操作进行统一AMQP是通过规定协议来统一数据交互的格式。
JMS限定了必须使用Java语言AMQP只是协议不规定实现方式因此是跨语言的。
JMS规定了两种消息模式而AMQP的消息模式更加丰富
1.4.4 市场常见消息队列
ActiveMQ基于JMS
ZeroMQ基于C语言开发
RabbitMQ基于AMQP协议erlang语言开发稳定性好
RocketMQ基于JMS阿里巴巴产品
Kafka类似MQ的产品分布式消息系统高吞吐量。
1.5 RabbitMQ的安装和配置
/usr/lwl/soft路径下建立rabbit目录
下面要安装的包就是rpm包
注
RMP 是 LINUX 下的一种软件的可执行程序你只要安装它就可以了。这种软件安装包通常是一个RPM包Redhat Linux Packet Manager就是Redhat的包管理器后缀是.rpm。
常用命令组合
ivh安装显示安装进度--install--verbose--hash
Uvh升级软件包--Update
qpl列出RPM软件包内的文件信息[Query Package list]
qpi列出RPM软件包的描述信息[Query Package install package(s)]
qf查找指定文件属于哪个RPM软件包[Query File]
Va校验所有的RPM软件包查找丢失的文件[View Lost]
e删除包
1、安装依赖环境
在rabbitmq目录下进行即可
yum -y install build-essential openssl openssl-devel unixODBC unixODBC-devel make gcc gcc-c kernel-devel m4 ncurses-devel tk tc xz
2、安装Erlang
rpm -ivh erlang-22.0.7-1.el7.x86_64.rpm
3、安装RabbitMQ
安装依赖的包
rpm -ivh socat-1.7.3.2-2.el7.x86_64.rpm
安装RabbitMQ
rpm -ivh rabbitmq-server-3.7.18-1.el7.noarch.rpm
4、RabbitMQ服务命令
systemctl start rabbitmq-server # 启动服务
systemctl stop rabbitmq-server # 停止服务
systemctl restart rabbitmq-server # 重启服务
systemctl status rabbitmq-server #查看状态
可以先启动服务
5、开启管理界面(图形化)
rabbitmq-plugins enable rabbitmq_management 6、修改默认配置信息
vim /usr/lib/rabbitmq/lib/rabbitmq_server-3.7.18/ebin/rabbit.app
# 比如修改密码、配置等等例如loopback_users中的 guest,只保留guest
修改前
39 {loopback_users, [guest]},
修改后39 {loopback_users, guest},
修改后可以重启一下RabbitMQ
systemctl restart rabbitmq-server
7、打开客户端
浏览器地址为
http://192.168.111.127:15672/
192.168.111.127是主机的IP地址
156721代表着客户端5672是默认端口号
初始用户和密码都是guest
登陆后出现下图即为安装成功 1.6 使用图形化添加一个用户和虚拟主机
1.6.1 添加一个用户 1.6.2 添加一个虚拟主机 1.6.3 配置用户和虚拟主机
这里是将新建的/lwl虚拟主机赋权给lwl用户 1.6.4 删除一个虚拟主机 1.6.5 使用可视化完成消息收发
创建一个队列在该队列中存放一条消息再接收一下这条消息 1.7 java测试RabbitMQ的工作模式
这里使用maven项目或者是SpringBoot项目都可以
这里使用的是maven项目后面有使用SpringBoot项目的示例创建项目时一般是父模块用来加载pom文件
子模块用来进行编写代码子模块会继承父模块的pom文件
父模块 testRabbitMQ
子模块1 rabbitprocedure
子模块2 rabbitconsumer
父模块添加依赖
dependencygroupIdcom.rabbitmq/groupIdartifactIdamqp-client/artifactIdversion5.6.0/version
/dependency
1.7.1 简单模式(Hello world)
生产者
package com.aaa.simple;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*** 简单模式生产者1*/
public class Procedure1 {public static void main(String[] args) throws IOException, TimeoutException {//1、创建工厂ConnectionFactory conn new ConnectionFactory();//2、设置参数conn.setHost(192.168.111.127); //设置ip地址默认为localhostconn.setPort(5672); //设置端口号默认为5672conn.setUsername(lwl); //设置rabbit登录用户名默认为guestconn.setPassword(密码); //设置rabbit登录密码默认为guestconn.setVirtualHost(/lwl); //设置虚拟主机,默认为 /
//3、需要通过管道建立连接 , 向上抛出异常Connection connection conn.newConnection();//4、创建channelChannel channel connection.createChannel();
//5、声明队列/** 查看方法中的参数按着ctrl点击下面的queueDeclare()会进入到接口中然后在进入到对应的实现类中* (String queue, boolean durable, boolean exclusive, boolean autoDelete, MapString, Object arguments)* queue队列的名字* durable是否需要持久化* exclusive是否独占(一个消费者对应一个生产者)* autoDelete是否自动删除* arguments对应的参数*/channel.queueDeclare(testrabbit,false,false,true,null);//6、发布消息/*** (String exchange, String routingKey, BasicProperties props, byte[] body)* exchange交换机的名字* routingKey绑定的key的名字路由名称,在简单模式下需要和队列的名字一致* props交换机的属性* body要发布的消息内容的byte数组*/byte[] msg hello rabbit!.getBytes();channel.basicPublish(,testrabbit,null,msg);}
}
运行完生产者之后可以在可视化界面看一下相应的队列中是否有对应的消息
消费者
package com.aaa.simple;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*** 简单模式消费者1*/
public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {//1、创建工厂ConnectionFactory conn new ConnectionFactory();//2、设置参数conn.setHost(192.168.111.127); //设置ip地址默认为localhostconn.setPort(5672); //设置端口号默认为5672conn.setUsername(lwl); //设置rabbit登录用户名默认为guestconn.setPassword(密码); //设置rabbit登录密码默认为guestconn.setVirtualHost(/lwl); //设置虚拟主机,默认为 /
//3、需要通过管道建立连接 , 向上抛出异常Connection connection conn.newConnection();//4、创建channelChannel channel connection.createChannel();//5、创建consumer对象consumer是接口所以这里需要使用匿名函数Consumer consumer new DefaultConsumer(channel){/*** 重写里面的handleDelivery方法* consumerTag备注的一些信息* envelope封装的信息比如交换机、路由等* properties协议的属性* body获取到的队列里面的消息字节数组形式*/Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println(consumerTag consumerTag);System.out.println(envelope envelope.getExchange());System.out.println(envelope envelope.getRoutingKey());System.out.println(properties properties);System.out.println(body new String(body));}};
//6、消费队列里面的消息/*** basicConsume有很多中构造方法这里使用的是有以下三个参数的构造方法* (String queue, boolean autoAck, Consumer callback)* queue代表队列的名字 刚才定义的队列名字为test* autoAck是否自动确认* callbackconsumer对象*/channel.basicConsume(test,true,consumer);
/*** consumerTag amq.ctag-r2fL0LOWbFZoXFhzvRUG2w* envelope * envelope test* properties #contentHeaderbasic(content-typenull, content-encodingnull, headers{}, delivery-mode1, prioritynull, correlation-idnull, reply-tonull, expirationnull, message-idnull, timestampnull, typenull, user-idnull, app-idnull, cluster-idnull)* body 测试发布消息*/}
}
1.7.2 工作队列模式(Work queues)
Work Queues与入门程序的简单模式相比多了一个或一些消费端多个消费端共同消费同一个队列中的消息。
应用场景对于 任务过重或任务较多情况使用工作队列可以提高任务处理的速度。 1.7.2.1轮询分发机制案例
生产者
package com.aaa.workqueue;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*** 工作队列模式生产者1*/
public class Procedure1 {public static void main(String[] args) throws IOException, TimeoutException {//1、创建工厂ConnectionFactory conn new ConnectionFactory();//2、设置参数conn.setHost(192.168.111.127); //设置ip地址默认为localhostconn.setPort(5672); //设置端口号默认为5672conn.setUsername(lwl); //设置rabbit登录用户名默认为guestconn.setPassword(密码); //设置rabbit登录密码默认为guestconn.setVirtualHost(/lwl); //设置虚拟主机,默认为 /
//3、需要通过管道建立连接 , 向上抛出异常Connection connection conn.newConnection();//4、创建channelChannel channel connection.createChannel();
//5、声明队列/** 查看方法中的参数按着ctrl点击下面的queueDeclare()会进入到接口中然后在进入到对应的实现类中* (String queue, boolean durable, boolean exclusive, boolean autoDelete, MapString, Object arguments)* queue队列的名字* durable是否需要持久化* exclusive是否独占(一个消费者对应一个生产者)* autoDelete是否自动删除* arguments对应的参数*/channel.queueDeclare(workqueue,false,false,true,null);//6、发布消息/*** (String exchange, String routingKey, BasicProperties props, byte[] body)* exchange交换机的名字* routingKey绑定的key的名字路由名称,在简单模式下需要和队列的名字一致* props交换机的属性* body要发布的消息内容的byte数组*/for (int i 0; i 10; i) {channel.basicPublish(,workqueue,null,(worki).getBytes());}
}
}
消费者1
package com.aaa.workqueue;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
/*** 工作队列模式消费者1*/
public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {//1、创建工厂ConnectionFactory conn new ConnectionFactory();//2、设置参数conn.setHost(192.168.111.127); //设置ip地址默认为localhostconn.setPort(5672); //设置端口号默认为5672conn.setUsername(lwl); //设置rabbit登录用户名默认为guestconn.setPassword(密码); //设置rabbit登录密码默认为guestconn.setVirtualHost(/lwl); //设置虚拟主机,默认为 /
//3、需要通过管道建立连接 , 向上抛出异常Connection connection conn.newConnection();//4、创建channelchannel.basicQos(1);//5、创建consumer对象consumer是接口所以这里需要使用匿名函数Consumer consumer new DefaultConsumer(channel){/*** 重写里面的handleDelivery方法* consumerTag备注的一些信息* envelope封装的信息比如交换机、路由等* properties协议的属性* body获取到的队列里面的消息字节数组形式*/Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println(body new String(body));
}};
//6、消费队列里面的消息/*** basicConsume有很多中构造方法这里使用的是有以下三个参数的构造方法* (String queue, boolean autoAck, Consumer callback)* queue代表队列的名字 刚才定义的队列名字为test* autoAck是否自动确认* callbackconsumer对象*/channel.basicConsume(workqueue,true,consumer);}
}
消费者2
package com.aaa.workqueue;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*** 工作队列模式消费者2*/
public class Consumer2 {public static void main(String[] args) throws IOException, TimeoutException {//1、创建工厂ConnectionFactory conn new ConnectionFactory();//2、设置参数conn.setHost(192.168.111.127); //设置ip地址默认为localhostconn.setPort(5672); //设置端口号默认为5672conn.setUsername(lwl); //设置rabbit登录用户名默认为guestconn.setPassword(密码); //设置rabbit登录密码默认为guestconn.setVirtualHost(/lwl); //设置虚拟主机,默认为 /
//3、需要通过管道建立连接 , 向上抛出异常Connection connection conn.newConnection();//4、创建channelChannel channel connection.createChannel();//5、创建consumer对象consumer是接口所以这里需要使用匿名函数Consumer consumer new DefaultConsumer(channel){/*** 重写里面的handleDelivery方法* consumerTag备注的一些信息* envelope封装的信息比如交换机、路由等* properties协议的属性* body获取到的队列里面的消息字节数组形式*/Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println(body new String(body));
}};//6、消费队列里面的消息/*** basicConsume有很多中构造方法这里使用的是有以下三个参数的构造方法* (String queue, boolean autoAck, Consumer callback)* queue代表队列的名字 刚才定义的队列名字为test* autoAck是否自动确认* callbackconsumer对象*/channel.basicConsume(workqueue,true,consumer);}
}
上面这种模式多个消费者是轮询分发的机制多个消费者一定会自动平均分配消息队列中的消息
而为什么会自动平均分配呢
消息自动确认因为我们在消费者消费的时候有一个参数设置为 autoAck:true,我们设置消费者接收消息自动确认而一般都是队列中分配好了那个消费者要传递什么信息直接一次全部传递过去不是消费一个确认一下。消费者接收到所有消息之后自动确认队列中会标记删除所有的信息他不关心你接收完信息之后的后续业务操作。就是只关心你是否收到了数据。消息自动确认机制完成一项任务可能需要几秒钟甚至几分钟如果一个消费者开始了一项长期的任务却只完成了一部分就挂了那么RabbitMQ一旦将消息传递给消费者就会立刻标记删除那么因为消费者挂了接收数据的时候已经确认应答了队列中的数据也删除了所以剩余接收到的信息也没了。
如果消息手动确认如果生产者发送10条消息消费者1拿到5条消费者2拿到5条不进行自动应答服务器队列的数据即使消费了我们没有应答就不会被标记删除保证服务器队列中的数据一直还在。如果消费者处理完了这条数据那么手动确认队列中知道已经确认了进行删除1、消费消息的时候参数设置为false
channel.basicConsume(workqueue,false,consumer);
2、我们在 handlerDeliver 方法中也需要设置手动确认 // 手动确认channel.basicAck(envelope.getDeliveryTag(), false);
我们设置一个信道每次只能消费一个消息如果其中一个消费者服务器挂掉了连接就断掉了剩余的未被手动确认的数据还在队列中保存。也能及时得把剩余的消息继续交给消费者2进行处理不耽误业务的持续进行。这就是能者多劳的机制。就是说处理快的消费者处理完业务会很快的手动确认然后再次进行接收新的消息处理慢的消费者经过一段时间处理之后再进行确认就会能者多劳业务处理快的接受的消息多处理满的接受的少
3、在消费消息之前设置信道中接收消息只能是1个chanel.basicQos(1); // 设置信道中一次只能消费一个信息
这样就能实现能者多劳机制
1.7.2.2能者多劳机制案例
生产者
package com.aaa.workqueue;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*** 工作队列模式生产者1*/
public class Procedure1 {public static void main(String[] args) throws IOException, TimeoutException {//1、创建工厂ConnectionFactory conn new ConnectionFactory();//2、设置参数conn.setHost(192.168.111.127); //设置ip地址默认为localhostconn.setPort(5672); //设置端口号默认为5672conn.setUsername(lwl); //设置rabbit登录用户名默认为guestconn.setPassword(密码); //设置rabbit登录密码默认为guestconn.setVirtualHost(/lwl); //设置虚拟主机,默认为 /
//3、需要通过管道建立连接 , 向上抛出异常Connection connection conn.newConnection();//4、创建channelChannel channel connection.createChannel();
//5、声明队列/** 查看方法中的参数按着ctrl点击下面的queueDeclare()会进入到接口中然后在进入到对应的实现类中* (String queue, boolean durable, boolean exclusive, boolean autoDelete, MapString, Object arguments)* queue队列的名字* durable是否需要持久化* exclusive是否独占(一个消费者对应一个生产者)* autoDelete是否自动删除* arguments对应的参数*/channel.queueDeclare(workqueue,false,false,true,null);//6、发布消息/*** (String exchange, String routingKey, BasicProperties props, byte[] body)* exchange交换机的名字* routingKey绑定的key的名字路由名称,在简单模式下需要和队列的名字一致* props交换机的属性* body要发布的消息内容的byte数组*/for (int i 0; i 10; i) {channel.basicPublish(,workqueue,null,(worki).getBytes());}
}
}
消费者1
package com.aaa.workqueue;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
/*** 工作队列模式消费者1*/
public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {//1、创建工厂ConnectionFactory conn new ConnectionFactory();//2、设置参数conn.setHost(192.168.111.127); //设置ip地址默认为localhostconn.setPort(5672); //设置端口号默认为5672conn.setUsername(lwl); //设置rabbit登录用户名默认为guestconn.setPassword(密码); //设置rabbit登录密码默认为guestconn.setVirtualHost(/lwl); //设置虚拟主机,默认为 /
//3、需要通过管道建立连接 , 向上抛出异常Connection connection conn.newConnection();//4、创建channelChannel channel connection.createChannel();// 设置信道中一次只能消费一个信息channel.basicQos(1);//5、创建consumer对象consumer是接口所以这里需要使用匿名函数Consumer consumer new DefaultConsumer(channel){/*** 重写里面的handleDelivery方法* consumerTag备注的一些信息* envelope封装的信息比如交换机、路由等* properties协议的属性* body获取到的队列里面的消息字节数组形式*/Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println(body new String(body));//让消费者1休眠try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}// 消息idmq 在 channel 中用来标识消息的 id可用于确认消息已接收long deliveryTag envelope.getDeliveryTag();/*** param1deliveryTag:用来标识消息的id* param2multiple是否批量。true:将一次性 ACK 所有小于 deliveryTag 的消息*/// 手动确认channel.basicAck(deliveryTag, false);}};
//6、消费队列里面的消息/*** basicConsume有很多中构造方法这里使用的是有以下三个参数的构造方法* (String queue, boolean autoAck, Consumer callback)* queue代表队列的名字 刚才定义的队列名字为test* autoAck是否自动确认* callbackconsumer对象*/channel.basicConsume(workqueue,false,consumer);}
}
消费者2
package com.aaa.workqueue;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*** 工作队列模式消费者2*/
public class Consumer2 {public static void main(String[] args) throws IOException, TimeoutException {//1、创建工厂ConnectionFactory conn new ConnectionFactory();//2、设置参数conn.setHost(192.168.111.127); //设置ip地址默认为localhostconn.setPort(5672); //设置端口号默认为5672conn.setUsername(lwl); //设置rabbit登录用户名默认为guestconn.setPassword(密码); //设置rabbit登录密码默认为guestconn.setVirtualHost(/lwl); //设置虚拟主机,默认为 /
//3、需要通过管道建立连接 , 向上抛出异常Connection connection conn.newConnection();//4、创建channelChannel channel connection.createChannel();// 设置信道中一次只能消费一个信息channel.basicQos(1);//5、创建consumer对象consumer是接口所以这里需要使用匿名函数Consumer consumer new DefaultConsumer(channel){/*** 重写里面的handleDelivery方法* consumerTag备注的一些信息* envelope封装的信息比如交换机、路由等* properties协议的属性* body获取到的队列里面的消息字节数组形式*/Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println(body new String(body));
// 消息idmq 在 channel 中用来标识消息的 id可用于确认消息已接收long deliveryTag envelope.getDeliveryTag();/*** param1deliveryTag:用来标识消息的id* param2multiple是否批量。true:将一次性 ACK 所有小于 deliveryTag 的消息*/// 手动确认channel.basicAck(deliveryTag, false);}};//6、消费队列里面的消息/*** basicConsume有很多中构造方法这里使用的是有以下三个参数的构造方法* (String queue, boolean autoAck, Consumer callback)* queue代表队列的名字 刚才定义的队列名字为test* autoAck是否自动确认* callbackconsumer对象*/channel.basicConsume(workqueue,false,consumer);
}
}
因为使用了手动提交而且每次队列中只有一条消息所以消息不再是平分
没有睡眠的消费者2效率会更高消费的消息会比消费者1多
1.7.3 发布订阅模式(Publish/Subscribe)
前面案例中只有3个角色
- P生产者也就是要发送消息的程序
- C消费者消息的接受者会一直等待消息到来。
- queue消息队列图中红色部分 而在订阅模型中多了一个exchange角色而且过程略有变化
生产者发消息给交换机交换机将消息路由分发给队列消费者监听队列接收信息
• P生产者也就是要发送消息的程序但是不再发送到队列中而是发给X交换机
• C消费者消息的接受者会一直等待消息到来。
• Queue消息队列接收消息、缓存消息。
• Exchange交换机图中的X。一方面接收生产者发送的消息。另一方面知道如何处理消息例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作取决于Exchange的类型。Exchange有常见以下3种类型◦ Fanout广播将消息交给所有绑定到交换机的队列◦ Direct定向把消息交给符合指定routing key 的队列◦ Topic通配符把消息交给符合routing pattern路由模式 的队列
Exchange交换机只负责转发消息不具备存储消息的能力因此如果没有任何队列与Exchange绑定或者没有符合路由规则的队列那么消息会丢失
生产者
package com.aaa.workqueue;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*** 发布订阅模式生产者1*/
public class Procedure1 {public static void main(String[] args) throws IOException, TimeoutException {//1、创建工厂ConnectionFactory conn new ConnectionFactory();//2、设置参数conn.setHost(192.168.111.127); //设置ip地址默认为localhostconn.setPort(5672); //设置端口号默认为5672conn.setUsername(lwl); //设置rabbit登录用户名默认为guestconn.setPassword(密码); //设置rabbit登录密码默认为guestconn.setVirtualHost(/lwl); //设置虚拟主机,默认为 /
//3、需要通过管道建立连接 , 向上抛出异常Connection connection conn.newConnection();//4、创建channelChannel channel connection.createChannel();
//5、声明一个交换机String exchangeName ex_fan_out;/*** (String exchange, BuiltinExchangeType type, boolean durable)* exchange交换机的名字* type交换机的名字* durable是否需要持久化*/channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT,false);
//6、声明两个队列/*** (String queue, boolean durable, boolean exclusive, boolean autoDelete, MapString, Object arguments)* queue队列的名字* durable是否需要持久化* exclusive是否独占(一个消费者对应一个生产者)* autoDelete是否自动删除* arguments对应的参数*/String queueName1 ex_fan_out_queue1;String queueName2 ex_fan_out_queue2;channel.queueDeclare(queueName1,false,false,true,null);channel.queueDeclare(queueName2,false,false,true,null);
//7、将队列和交换机进行绑定/** queueBind 和 exchangeBind是一样的使用*(String queue, String exchange, String routingKey)* queue队列的名字* exchange交换机的名字* routingKey路由名称广播模式下不需要设置为即可*/channel.queueBind(queueName1,exchangeName,);channel.queueBind(queueName2,exchangeName,);
//6、发布消息/*** (String exchange, String routingKey, BasicProperties props, byte[] body)* exchange交换机的名字* routingKey绑定的key的名字路由名称,在简单模式下需要和队列的名字一致* props交换机的属性* body要发布的消息内容的byte数组*/channel.basicPublish(exchangeName,,null,exchange_queue.getBytes());}
}
消费者1
package com.aaa.publish;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*** 发布订阅模式消费者1*/
public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {//1、创建工厂ConnectionFactory conn new ConnectionFactory();//2、设置参数conn.setHost(192.168.111.127); //设置ip地址默认为localhostconn.setPort(5672); //设置端口号默认为5672conn.setUsername(lwl); //设置rabbit登录用户名默认为guestconn.setPassword(密码); //设置rabbit登录密码默认为guestconn.setVirtualHost(/lwl); //设置虚拟主机,默认为 /
//3、需要通过管道建立连接 , 向上抛出异常Connection connection conn.newConnection();//4、创建channelChannel channel connection.createChannel();//5、创建consumer对象consumer是接口所以这里需要使用匿名函数Consumer consumer new DefaultConsumer(channel){/*** 重写里面的handleDelivery方法* consumerTag备注的一些信息* envelope封装的信息比如交换机、路由等* properties协议的属性* body获取到的队列里面的消息字节数组形式*/Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println(body new String(body));}};
//6、消费队列里面的消息/*** basicConsume有很多中构造方法这里使用的是有以下三个参数的构造方法* (String queue, boolean autoAck, Consumer callback)* queue代表队列的名字 刚才定义的队列名字为test* autoAck是否自动确认* callbackconsumer对象*/String queueName1 ex_fan_out_queue1;channel.basicConsume(queueName1,true,consumer);}
}
消费者2
package com.aaa.publish;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*** 发布订阅模式消费者2*/
public class Consumer2 {public static void main(String[] args) throws IOException, TimeoutException {//1、创建工厂ConnectionFactory conn new ConnectionFactory();//2、设置参数conn.setHost(192.168.111.127); //设置ip地址默认为localhostconn.setPort(5672); //设置端口号默认为5672conn.setUsername(lwl); //设置rabbit登录用户名默认为guestconn.setPassword(密码); //设置rabbit登录密码默认为guestconn.setVirtualHost(/lwl); //设置虚拟主机,默认为 /
//3、需要通过管道建立连接 , 向上抛出异常Connection connection conn.newConnection();//4、创建channelChannel channel connection.createChannel();//5、创建consumer对象consumer是接口所以这里需要使用匿名函数Consumer consumer new DefaultConsumer(channel){/*** 重写里面的handleDelivery方法* consumerTag备注的一些信息* envelope封装的信息比如交换机、路由等* properties协议的属性* body获取到的队列里面的消息字节数组形式*/Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println(body new String(body));}};
//6、消费队列里面的消息/*** basicConsume有很多中构造方法这里使用的是有以下三个参数的构造方法* (String queue, boolean autoAck, Consumer callback)* queue代表队列的名字 刚才定义的队列名字为test* autoAck是否自动确认* callbackconsumer对象*/String queueName2 ex_fan_out_queue2;channel.basicConsume(queueName2,true,consumer);}
}
因为生产者发布的消息中routingkey是也就是广播模式
两个队列都能匹配上
所以同样的消息两个消费者都能消费
1.7.4 路由模式(Routing)
路由模式特点
• 队列与交换机的绑定不能是任意绑定了而是要指定一个RoutingKey路由key
• 消息的发送方在 向Exchange发送消息时也必须指定消息的 RoutingKey。
• Exchange不再把消息交给每一个绑定的队列而是根据消息的Routing Key进行判断只有队列的Routingkey与消息的 Routing key完全一致才会接收到消息 • P生产者向Exchange发送消息发送消息时会指定一个routing key。
• XExchange交换机接收生产者的消息然后把消息递交给 与routing key完全匹配的队列
• C1消费者其所在队列指定了需要routing key 为 error 的消息
• C2消费者其所在队列指定了需要routing key 为 info、error、warning 的消息
生产者
package com.aaa.routing;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*** 路由模式生产者1*/
public class Procedure1 {public static void main(String[] args) throws IOException, TimeoutException {//1、创建工厂ConnectionFactory conn new ConnectionFactory();//2、设置参数conn.setHost(192.168.111.127); //设置ip地址默认为localhostconn.setPort(5672); //设置端口号默认为5672conn.setUsername(lwl); //设置rabbit登录用户名默认为guestconn.setPassword(密码); //设置rabbit登录密码默认为guestconn.setVirtualHost(/lwl); //设置虚拟主机,默认为 /
//3、需要通过管道建立连接 , 向上抛出异常Connection connection conn.newConnection();//4、创建channelChannel channel connection.createChannel();
//5、声明一个交换机String exchangeName ex_dir;/*** (String exchange, BuiltinExchangeType type, boolean durable)* exchange交换机的名字* type交换机的名字* durable是否需要持久化*/channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT,false);
//6、声明两个队列/*** (String queue, boolean durable, boolean exclusive, boolean autoDelete, MapString, Object arguments)* queue队列的名字* durable是否需要持久化* exclusive是否独占(一个消费者对应一个生产者)* autoDelete是否自动删除* arguments对应的参数*/String queueName1 ex_dir_queue1;String queueName2 ex_dir_queue2;channel.queueDeclare(queueName1,false,false,true,null);channel.queueDeclare(queueName2,false,false,true,null);
//7、将队列和交换机进行绑定/** queueBind 和 exchangeBind是一样的使用*(String queue, String exchange, String routingKey)* queue队列的名字* exchange交换机的名字* routingKey路由名称路由模式下需要定义*/channel.queueBind(queueName1,exchangeName,error);
channel.queueBind(queueName2,exchangeName,error);channel.queueBind(queueName2,exchangeName,info);channel.queueBind(queueName2,exchangeName,warning);
//6、发布消息/*** (String exchange, String routingKey, BasicProperties props, byte[] body)* exchange交换机的名字* routingKey绑定的key的名字路由名称,在简单模式下需要和队列的名字一致* props交换机的属性* body要发布的消息内容的byte数组*///如果这里routingKey定义为error那么按照绑定路由和交换机的规则两个消费者都能收到消息//如果这里routingKey定义为warning那么按照绑定路由和交换机的规则只有消费者2能收到消息channel.basicPublish(exchangeName,info,null,exchange_dir_queue.getBytes());}
}
消费者1
package com.aaa.routing;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*** 路由模式消费者1*/
public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {//1、创建工厂ConnectionFactory conn new ConnectionFactory();//2、设置参数conn.setHost(192.168.111.127); //设置ip地址默认为localhostconn.setPort(5672); //设置端口号默认为5672conn.setUsername(lwl); //设置rabbit登录用户名默认为guestconn.setPassword(密码); //设置rabbit登录密码默认为guestconn.setVirtualHost(/lwl); //设置虚拟主机,默认为 /
//3、需要通过管道建立连接 , 向上抛出异常Connection connection conn.newConnection();//4、创建channelChannel channel connection.createChannel();//5、创建consumer对象consumer是接口所以这里需要使用匿名函数Consumer consumer new DefaultConsumer(channel){/*** 重写里面的handleDelivery方法* consumerTag备注的一些信息* envelope封装的信息比如交换机、路由等* properties协议的属性* body获取到的队列里面的消息字节数组形式*/Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println(body new String(body));}};
//6、消费队列里面的消息/*** basicConsume有很多中构造方法这里使用的是有以下三个参数的构造方法* (String queue, boolean autoAck, Consumer callback)* queue代表队列的名字 刚才定义的队列名字为test* autoAck是否自动确认* callbackconsumer对象*/String queueName1 ex_dir_queue1;channel.basicConsume(queueName1,true,consumer);}
}
消费者2
package com.aaa.routing;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*** 路由模式消费者2*/
public class Consumer2 {public static void main(String[] args) throws IOException, TimeoutException {//1、创建工厂ConnectionFactory conn new ConnectionFactory();//2、设置参数conn.setHost(192.168.111.127); //设置ip地址默认为localhostconn.setPort(5672); //设置端口号默认为5672conn.setUsername(lwl); //设置rabbit登录用户名默认为guestconn.setPassword(密码); //设置rabbit登录密码默认为guestconn.setVirtualHost(/lwl); //设置虚拟主机,默认为 /
//3、需要通过管道建立连接 , 向上抛出异常Connection connection conn.newConnection();//4、创建channelChannel channel connection.createChannel();//5、创建consumer对象consumer是接口所以这里需要使用匿名函数Consumer consumer new DefaultConsumer(channel){/*** 重写里面的handleDelivery方法* consumerTag备注的一些信息* envelope封装的信息比如交换机、路由等* properties协议的属性* body获取到的队列里面的消息字节数组形式*/Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println(body new String(body));}};
//6、消费队列里面的消息/*** basicConsume有很多中构造方法这里使用的是有以下三个参数的构造方法* (String queue, boolean autoAck, Consumer callback)* queue代表队列的名字 刚才定义的队列名字为test* autoAck是否自动确认* callbackconsumer对象*/String queueName2 ex_dir_queue2;channel.basicConsume(queueName2,true,consumer);}
}
因为生产者发布的消息中routingkey是info
所以只有队列ex_dir_queue2能匹配上
所以只有消费者2能消费消息
1.7.5 通配符模式(Topics)
Topic类型与Direct相比都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符
Routingkey 一般都是有一个或多个单词组成多个单词之间以”.”分割例如 item.insert
通配符规则
#匹配一个或多个词
*匹配一个词
举例
item.#能够匹配item.insert.abc 或者 item.insert
item.*只能匹配item.insert 或者 item.abc • 红色Queue绑定的是usa.# 因此凡是以 usa.开头的routing key 都会被匹配到
• 黄色Queue绑定的是#.news 因此凡是以 .news结尾的 routing key 都会被匹配
生产者
package com.aaa.topics;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*** 主题(通配符)模式生产者1*/
public class Procedure1 {public static void main(String[] args) throws IOException, TimeoutException {//1、创建工厂ConnectionFactory conn new ConnectionFactory();//2、设置参数conn.setHost(192.168.111.127); //设置ip地址默认为localhostconn.setPort(5672); //设置端口号默认为5672conn.setUsername(lwl); //设置rabbit登录用户名默认为guestconn.setPassword(密码); //设置rabbit登录密码默认为guestconn.setVirtualHost(/lwl); //设置虚拟主机,默认为 /
//3、需要通过管道建立连接 , 向上抛出异常Connection connection conn.newConnection();//4、创建channelChannel channel connection.createChannel();
//5、声明一个交换机String exchangeName ex_topic;/*** (String exchange, BuiltinExchangeType type, boolean durable)* exchange交换机的名字* type交换机的名字* durable是否需要持久化*/channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC,false);
//6、声明两个队列/*** (String queue, boolean durable, boolean exclusive, boolean autoDelete, MapString, Object arguments)* queue队列的名字* durable是否需要持久化* exclusive是否独占(一个消费者对应一个生产者)* autoDelete是否自动删除* arguments对应的参数*/String queueName1 ex_topic_queue1;String queueName2 ex_topic_queue2;channel.queueDeclare(queueName1,false,false,true,null);channel.queueDeclare(queueName2,false,false,true,null);
//7、将队列和交换机进行绑定/** queueBind 和 exchangeBind是一样的使用*(String queue, String exchange, String routingKey)* queue队列的名字* exchange交换机的名字* routingKey路由名称主题模式下需要定义的有通配符*/channel.queueBind(queueName1,exchangeName,*.error);
channel.queueBind(queueName2,exchangeName,#.error);channel.queueBind(queueName2,exchangeName,#.info);channel.queueBind(queueName2,exchangeName,#.warning);
//6、发布消息/*** (String exchange, String routingKey, BasicProperties props, byte[] body)* exchange交换机的名字* routingKey绑定的key的名字路由名称,在简单模式下需要和队列的名字一致* props交换机的属性* body要发布的消息内容的byte数组*///发布消息的时候不能有通配符channel.basicPublish(exchangeName,test.aaa.error,null,exchange_topic_queue.getBytes());}
}
消费者1
package com.aaa.topics;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*** 主题模式消费者1*/
public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {//1、创建工厂ConnectionFactory conn new ConnectionFactory();//2、设置参数conn.setHost(192.168.111.127); //设置ip地址默认为localhostconn.setPort(5672); //设置端口号默认为5672conn.setUsername(lwl); //设置rabbit登录用户名默认为guestconn.setPassword(密码); //设置rabbit登录密码默认为guestconn.setVirtualHost(/lwl); //设置虚拟主机,默认为 /
//3、需要通过管道建立连接 , 向上抛出异常Connection connection conn.newConnection();//4、创建channelChannel channel connection.createChannel();//5、创建consumer对象consumer是接口所以这里需要使用匿名函数Consumer consumer new DefaultConsumer(channel){/*** 重写里面的handleDelivery方法* consumerTag备注的一些信息* envelope封装的信息比如交换机、路由等* properties协议的属性* body获取到的队列里面的消息字节数组形式*/Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println(body new String(body));}};
//6、消费队列里面的消息/*** basicConsume有很多中构造方法这里使用的是有以下三个参数的构造方法* (String queue, boolean autoAck, Consumer callback)* queue代表队列的名字 刚才定义的队列名字为test* autoAck是否自动确认* callbackconsumer对象*/String queueName1 ex_topic_queue1;String queueName2 ex_topic_queue2;channel.basicConsume(queueName1,true,consumer);}
}
消费者2
package com.aaa.topics;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*** 主题模式消费者2*/
public class Consumer2 {public static void main(String[] args) throws IOException, TimeoutException {//1、创建工厂ConnectionFactory conn new ConnectionFactory();//2、设置参数conn.setHost(192.168.111.127); //设置ip地址默认为localhostconn.setPort(5672); //设置端口号默认为5672conn.setUsername(lwl); //设置rabbit登录用户名默认为guestconn.setPassword(密码); //设置rabbit登录密码默认为guestconn.setVirtualHost(/lwl); //设置虚拟主机,默认为 /
//3、需要通过管道建立连接 , 向上抛出异常Connection connection conn.newConnection();//4、创建channelChannel channel connection.createChannel();//5、创建consumer对象consumer是接口所以这里需要使用匿名函数Consumer consumer new DefaultConsumer(channel){/*** 重写里面的handleDelivery方法* consumerTag备注的一些信息* envelope封装的信息比如交换机、路由等* properties协议的属性* body获取到的队列里面的消息字节数组形式*/Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println(body new String(body));}};
//6、消费队列里面的消息/*** basicConsume有很多中构造方法这里使用的是有以下三个参数的构造方法* (String queue, boolean autoAck, Consumer callback)* queue代表队列的名字 刚才定义的队列名字为test* autoAck是否自动确认* callbackconsumer对象*/String queueName1 ex_topic_queue1;String queueName2 ex_topic_queue2;channel.basicConsume(queueName2,true,consumer);}
}
因为生产者发布的消息中routingkey是test.aaa.error
所以只有队列ex_topic_queue2能匹配上
所以只有消费者2能消费消息
1.7.6 模式间的异同
1、简单模式 HelloWorld
一个生产者、一个消费者不需要设置交换机使用默认的交换机
2、工作队列模式 Work Queue
一个生产者、多个消费者竞争关系不需要设置交换机使用默认的交换机
3、发布订阅模式 Publish/subscribe
需要设置类型为fanout的交换机并且交换机和队列进行绑定当发送消息到交换机后交换机会将消息发送到绑定的队列
4、路由模式 Routing
需要设置类型为direct的交换机交换机和队列进行绑定并且指定routing key当发送消息到交换机后交换机会根据routing key将消息发送到对应的队列
5、通配符模式 Topic
需要设置类型为topic的交换机交换机和队列进行绑定并且指定通配符方式的routing key当发送消息到交换机后交换机会根据routing key将消息发送到对应的队列
tips
Topic主题模式可以实现 Publish/Subscribe发布与订阅模式 和 Routing路由模式 的功能只是Topic在配置routing key 的时候可以使用通配符显得更加灵活。
1.7.7 使用可视化实现主题模式
1、创建一个交换机 2、创建两个队列 3、绑定交换机和队列
①使用交换机绑定队列 ②使用队列绑定交换机 4、使用交换机发布消息 5、查看结果 1.8 SpringBoot项目整合RabbitMQ
1.8.1 介绍
在Spring项目中可以使用Spring-Rabbit去操作RabbitMQ
尤其是在spring boot项目中只需要引入对应的amqp启动器依赖即可方便的使用RabbitTemplate发送消息使用注解接收消息。
生产者工程
1. application.yml或者是application.properties文件配置相关信息
2. 在生产者工程中编写配置类用于创建交换机和队列并进行绑定
3. 注入RabbitTemplate对象通过RabbitTemplate对象发送消息到交换机
消费者工程
4. application.yml或者是application.properties文件配置相关信息
5. 创建消息处理类用于接收队列中的消息并进行处理
1.8.2 创建父工程
父工程一般是为了加载依赖创建一个SpringBoot项目
修改pom文件 !--1. 父工程依赖先注释原本的父工程依赖 --parentgroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-parent/artifactIdversion2.1.4.RELEASE/versionrelativePath//parent!-- 操作RabbitMQ的依赖 --dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency
修改配置文件(application.properties)
#主机的IP地址默认为localhost
spring.rabbitmq.host192.168.111.127
#虚拟主机的名字默认为/
spring.rabbitmq.virtual-host/lwl
#登录的用户名默认为guest
spring.rabbitmq.usernamelwl
#登录的密码默认为guest
spring.rabbitmq.password密码
#RabbitMQ的端口号默认为5672
spring.rabbitmq.port5672
1.8.3 搭建生产者工程
生产者工程和主工程可以建成父子项目
生产者工程可以maven项目或者是SpringBoot项目1.如果生产者工程是maven项目是没有application.properties文件和启动类文件的需要自己创建何编写其中application.properties是在resources目录下启动类文件是在java目录下的最外层包下2.如果是SpringBoot项目那么在对应的配置文件中编写即可
如果父类中pom文件没有amqp的依赖也可以在子项目中加上本项目已经在父项目中配置过 dependenciesdependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency/dependencies
1、配置文件
#主机的IP地址默认为localhost
spring.rabbitmq.host192.168.111.127
#虚拟主机的名字默认为/
spring.rabbitmq.virtual-host/lwl
#登录的用户名默认为guest
spring.rabbitmq.usernamelwl
#登录的密码默认为guest
spring.rabbitmq.password密码
#RabbitMQ的端口号默认为5672
spring.rabbitmq.port5672
#配置文件中避免有空格会被错误识别
2、启动类文件
/*** 启动类文件一般是以项目名字加Test进行命名的* SpringBootApplication:启动类注解* run方法中写的是当前类的class文件*/
SpringBootApplication
public class ProcedureRabbitTest {public static void main(String[] args) {SpringApplication.run(ProcedureRabbitTest.class);}
}
3、配置生产者交换机和队列
package com.aaa.procedure.test;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/*** Configuration:声明为配置类可以更早加载*/
Configuration
public class ProcedureConfig {//声明队列的名字private static final String ROUTING_QUEUE_NAME1 routing_queue_name1;private static final String ROUTING_QUEUE_NAME2 routing_queue_name2;
//声明一个交换机的名字private static final String ROUTING_EXCHANGE_NAME routing_exchange_name;
/*** 声明一个交换机并作为bean注入到spring容器中* durable默认为durable* autoDelete默认为true* 交换机起名字在绑定队列时使用*/Bean(name ex)public Exchange getExchange(){// 交换机构建器.交换机类型(交换机名字).是否持久化.是否自动删除.构建return ExchangeBuilder.directExchange(ROUTING_EXCHANGE_NAME).durable(false).autoDelete().build();}
/*** 声明一个队列并作为bean注入到spring容器中* autoDelete默认为true* 队列起名字在绑定交换机时使用*/Bean(name qu1)public Queue getQueue1(){// 队列构建器.不持久化(队列名字).自动删除.构建return QueueBuilder.nonDurable(ROUTING_QUEUE_NAME1).autoDelete().build();}Bean(name qu2)public Queue getQueue2(){// 队列构建器.不持久化(队列名字).自动删除.构建return QueueBuilder.nonDurable(ROUTING_QUEUE_NAME2).autoDelete().build();}
/*** 绑定队列和交换机后作为bean注入到容器中* Qualifier注解为了标明对应的交换机和队列*/Beanpublic Binding bingQu1AndEx(Qualifier(ex) Exchange exchange,Qualifier(qu1) Queue queue){// 绑定构建器.队列名字.交换机名字.routingkey.参数return BindingBuilder.bind(queue).to(exchange).with(error).noargs();}Beanpublic Binding bingQu2AndEx(Qualifier(ex) Exchange exchange,Qualifier(qu2) Queue queue){// 绑定构建器.队列名字.交换机名字.routingkey.参数return BindingBuilder.bind(queue).to(exchange).with(warnings).noargs();}
}
4、使用测试类发布消息
package com.aaa.procedure;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringRunner;
import javax.annotation.Resource;
/*** SpringBootTest:声明为测试类* 因为SpringBoot的版本较低所以还需要添加这两个注解来确保项目没有问题* RunWith(SpringRunner.class)* ContextConfiguration(classes {ProcedureRabbitTest.class})*/
SpringBootTest
RunWith(SpringRunner.class)
ContextConfiguration(classes ProcedureRabbitTest.class)
public class TestPublish {
//将操作mq的对象引入进来是通过jar包amqp进行封装的对象Resourceprivate RabbitTemplate rabbitTemplate;
Testpublic void testRouting(){/*** 使用convertAndSend发送消息* (String exchange, String routingKey, Object object)* exchange交换机的名字* routingKey定向路由* object消息的内容*/rabbitTemplate.convertAndSend(routing_exchange_name,warnings,测试SpringBoot路由模式生产者);}
}
1.8.4 搭建消费者工程
消费者工程和主工程可以建成父子项目
消费者工程可以maven项目或者是SpringBoot项目1.如果消费者工程是maven项目是没有application.properties文件和启动类文件的需要自己创建何编写其中application.properties是在resources目录下启动类文件是在java目录下的最外层包下2.如果是SpringBoot项目那么在对应的配置文件中编写即可
如果父类中pom文件没有amqp的依赖也可以在子项目中加上本项目已经在父项目中配置过 dependenciesdependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency/dependencies
1、配置文件
#主机的IP地址默认为localhost
spring.rabbitmq.host192.168.111.127
#虚拟主机的名字默认为/
spring.rabbitmq.virtual-host/lwl
#登录的用户名默认为guest
spring.rabbitmq.usernamelwl
#登录的密码默认为guest
spring.rabbitmq.password密码
#RabbitMQ的端口号默认为5672
spring.rabbitmq.port5672
#配置文件中避免有空格会被错误识别
2、启动类
/*** 启动类文件一般是以项目名字加Test进行命名的* SpringBootApplication:启动类注解* run方法中写的是当前类的class文件*/
SpringBootApplication
public class ConsumerRabbitTest {public static void main(String[] args) {SpringApplication.run(ConsumerRabbitTest.class);}
}
3、监听队列
package com.aaa.consumer.test;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/*** Component将该类作为注解bean注入* RabbitListener标注监听队列的名字*/
Component
public class ConsumerListener {
//监听队列里面的消息可以同时监听很多个队列RabbitListener(queues {routing_queue_name1,routing_queue_name2})public void listenerQueues(Message message){//取出队列里面的消息bytes类型并将其转换为String类型String msg new String(message.getBody());System.out.println(监听到的消息为msg);}
}
先运行上述测试程序交换机和队列才能先被声明和绑定然后启动消费者在消费者工程中控制台查看是否接收到对应消息。 文章转载自： http://www.morning.zdhnm.cn.gov.cn.zdhnm.cn http://www.morning.nytpt.cn.gov.cn.nytpt.cn http://www.morning.tbksk.cn.gov.cn.tbksk.cn http://www.morning.mrxqd.cn.gov.cn.mrxqd.cn http://www.morning.dspqc.cn.gov.cn.dspqc.cn http://www.morning.demoux.com.gov.cn.demoux.com http://www.morning.lsxabc.com.gov.cn.lsxabc.com http://www.morning.qrcsb.cn.gov.cn.qrcsb.cn http://www.morning.mxftp.com.gov.cn.mxftp.com http://www.morning.zwndt.cn.gov.cn.zwndt.cn http://www.morning.ychrn.cn.gov.cn.ychrn.cn http://www.morning.psdsk.cn.gov.cn.psdsk.cn http://www.morning.wrqw.cn.gov.cn.wrqw.cn http://www.morning.wjrtg.cn.gov.cn.wjrtg.cn http://www.morning.qhydkj.com.gov.cn.qhydkj.com http://www.morning.c7510.cn.gov.cn.c7510.cn http://www.morning.etsaf.com.gov.cn.etsaf.com http://www.morning.clndl.cn.gov.cn.clndl.cn http://www.morning.nwcgj.cn.gov.cn.nwcgj.cn http://www.morning.pamdeer.com.gov.cn.pamdeer.com http://www.morning.wknj.cn.gov.cn.wknj.cn http://www.morning.hphqy.cn.gov.cn.hphqy.cn http://www.morning.nngq.cn.gov.cn.nngq.cn http://www.morning.fdjwl.cn.gov.cn.fdjwl.cn http://www.morning.gnbtp.cn.gov.cn.gnbtp.cn http://www.morning.xhxsr.cn.gov.cn.xhxsr.cn http://www.morning.ffydh.cn.gov.cn.ffydh.cn http://www.morning.mnnxt.cn.gov.cn.mnnxt.cn http://www.morning.qgjxy.cn.gov.cn.qgjxy.cn http://www.morning.mrncd.cn.gov.cn.mrncd.cn http://www.morning.xdnhw.cn.gov.cn.xdnhw.cn http://www.morning.cgtrz.cn.gov.cn.cgtrz.cn http://www.morning.xprq.cn.gov.cn.xprq.cn http://www.morning.pclgj.cn.gov.cn.pclgj.cn http://www.morning.bzlfw.cn.gov.cn.bzlfw.cn http://www.morning.xxrgt.cn.gov.cn.xxrgt.cn http://www.morning.kghhl.cn.gov.cn.kghhl.cn http://www.morning.dfrenti.com.gov.cn.dfrenti.com http://www.morning.sfmqm.cn.gov.cn.sfmqm.cn http://www.morning.brsgw.cn.gov.cn.brsgw.cn http://www.morning.wjyyg.cn.gov.cn.wjyyg.cn http://www.morning.ndnhf.cn.gov.cn.ndnhf.cn http://www.morning.jtszm.cn.gov.cn.jtszm.cn http://www.morning.iznek.com.gov.cn.iznek.com http://www.morning.lqznq.cn.gov.cn.lqznq.cn http://www.morning.ygwyt.cn.gov.cn.ygwyt.cn http://www.morning.qkqgj.cn.gov.cn.qkqgj.cn http://www.morning.mkpqr.cn.gov.cn.mkpqr.cn http://www.morning.qkskm.cn.gov.cn.qkskm.cn http://www.morning.mqmxg.cn.gov.cn.mqmxg.cn http://www.morning.btlsb.cn.gov.cn.btlsb.cn http://www.morning.zqmdn.cn.gov.cn.zqmdn.cn http://www.morning.gtdf.cn.gov.cn.gtdf.cn http://www.morning.3dcb8231.cn.gov.cn.3dcb8231.cn http://www.morning.sxbgc.cn.gov.cn.sxbgc.cn http://www.morning.lgtzd.cn.gov.cn.lgtzd.cn http://www.morning.mdwlg.cn.gov.cn.mdwlg.cn http://www.morning.fmkbk.cn.gov.cn.fmkbk.cn http://www.morning.bfhrj.cn.gov.cn.bfhrj.cn http://www.morning.zplzj.cn.gov.cn.zplzj.cn http://www.morning.wyrsn.cn.gov.cn.wyrsn.cn http://www.morning.cbtn.cn.gov.cn.cbtn.cn http://www.morning.lfdzr.cn.gov.cn.lfdzr.cn http://www.morning.nbrdx.cn.gov.cn.nbrdx.cn http://www.morning.ljhnn.cn.gov.cn.ljhnn.cn http://www.morning.jypsm.cn.gov.cn.jypsm.cn http://www.morning.ybmp.cn.gov.cn.ybmp.cn http://www.morning.sffwz.cn.gov.cn.sffwz.cn http://www.morning.rjnrf.cn.gov.cn.rjnrf.cn http://www.morning.lrwsk.cn.gov.cn.lrwsk.cn http://www.morning.wmfr.cn.gov.cn.wmfr.cn http://www.morning.ggfdq.cn.gov.cn.ggfdq.cn http://www.morning.gcfrt.cn.gov.cn.gcfrt.cn http://www.morning.tsynj.cn.gov.cn.tsynj.cn http://www.morning.xbbrh.cn.gov.cn.xbbrh.cn http://www.morning.ywndg.cn.gov.cn.ywndg.cn http://www.morning.shyqcgw.cn.gov.cn.shyqcgw.cn http://www.morning.yuanshenglan.com.gov.cn.yuanshenglan.com http://www.morning.lhgkr.cn.gov.cn.lhgkr.cn http://www.morning.mnmrx.cn.gov.cn.mnmrx.cn
