1.介绍
1.1.RabbitMQ
MQ全称为Message Queue,即消息队列, RabbitMQ是由erlang语言开发,基于AMQP(Advanced Message
Queue 高级消息队列协议)协议实现的消息队列,它是一种应用程序之间的通信方法,消息队列在分布式系统开
发中应用非常广泛。RabbitMQ官方地址:添加链接描述
开发中消息队列通常有如下应用场景:
1、任务异步处理。
将不需要同步处理的并且耗时长的操作由消息队列通知消息接收方进行异步处理。提高了应用程序的响应时间。
2、应用程序解耦合
MQ相当于一个中介,生产方通过MQ与消费方交互,它将应用程序进行解耦合。
1.2.市场上还有哪些消息队列?
ActiveMQ,RabbitMQ,ZeroMQ,Kafka,MetaMQ,RocketMQ、Redis。
1.3.为什么使用RabbitMQ呢?
1、使得简单,功能强大。
2、基于AMQP协议。
3、社区活跃,文档完善。
4、高并发性能好,这主要得益于Erlang语言。
5、Spring Boot默认已集成RabbitMQ
2.快速入门
RabbitMQ的基本结构:
2.1.组成部分说明如下:
Broker :消息队列服务进程,此进程包括两个部分:Exchange和Queue。
Exchange :消息队列交换机,按一定的规则将消息路由转发到某个队列,对消息进行过虑。
Queue :消息队列,存储消息的队列,消息到达队列并转发给指定的消费方。
Producer :消息生产者,即生产方客户端,生产方客户端将消息发送到MQ。
Consumer :消息消费者,即消费方客户端,接收MQ转发的消息。
2.2.消息发布接收流程:
-----发送消息-----
1、生产者和Broker建立TCP连接。
2、生产者和Broker建立通道。
3、生产者通过通道消息发送给Broker,由Exchange将消息进行转发。
4、Exchange将消息转发到指定的Queue(队列)
----接收消息-----
1、消费者和Broker建立TCP连接
2、消费者和Broker建立通道
3、消费者监听指定的Queue(队列)
4、当有消息到达Queue时Broker默认将消息推送给消费者。
5、消费者接收到消息。
2.3.下载安装
RabbitMQ由Erlang语言开发,Erlang语言用于并发及分布式系统的开发,在电信领域应用广泛,OTP(Open
Telecom Platform)作为Erlang语言的一部分,包含了很多基于Erlang开发的中间件及工具库,安装RabbitMQ需
要安装Erlang/OTP,并保持版本匹配,如下图:
RabbitMQ的下载地址:http://www.rabbitmq.com/download.html
2.3.1.下载erlang
地址如下:http://erlang.org/download/otp_win64_20.3.exe
或去老师提供的软件包中找到 otp_win64_20.3.exe,以管理员方式运行此文件,安装。
erlang安装完成需要配置erlang环境变量: ERLANG_HOME=D:\Program Files\erl9.3 在path中添
加%ERLANG_HOME%\bin;
2.3.2.安装RabbitMQ
https://github.com/rabbitmq/rabbitmq-server/releases/tag/v3.7.3
2.4.启动
安装成功后会自动创建RabbitMQ服务并且启动。
2.4.1.从开始菜单启动RabbitMQ
完成在开始菜单找到RabbitMQ的菜单:
2.4.2.如果没有开始菜单则进入安装目录下sbin目录手动启动:
2.4.3.安装并运行服务
rabbitmq-service.bat install 安装服务 rabbitmq-service.bat stop 停止服务 rabbitmq-service.bat start 启动服务
2.4.4.安装管理插件
安装rabbitMQ的管理插件,方便在浏览器端管理RabbitMQ
管理员身份运行 rabbitmq-plugins.bat enable rabbitmq_management
2.4.5.启动成功 登录RabbitMQ
进入浏览器,输入:http://localhost:15672
初始账号和密码:guest/guest
2.4.6.注意事项
1、安装erlang和rabbitMQ以管理员身份运行。
2、当卸载重新安装时会出现RabbitMQ服务注册失败,此时需要进入注册表清理erlang
搜索RabbitMQ、ErlSrv,将对应的项全部删除。
2.5.Hello World
按照官方教程(http://www.rabbitmq.com/getstarted.html) 测试hello world:
2.5.1.搭建环境
1)java client
生产者和消费者都属于客户端,rabbitMQ的java客户端如下:
我们先用 rabbitMQ官方提供的java client测试,目的是对RabbitMQ的交互过程有个清晰的认识。
参考 :https://github.com/rabbitmq/rabbitmq-java-client/
2 )创建maven工程
创建生产者工程和消费者工程,分别加入RabbitMQ java client的依赖。
test-rabbitmq-producer:生产者工程
test-rabbitmq-consumer:消费者工程
<dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp‐client</artifactId> <version>4.0.3</version><!‐‐此版本与spring boot 1.5.9版本匹配‐‐> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring‐boot‐starter‐logging</artifactId> </dependency>
2.5.2.生产者
在生产者工程下的test中创建测试类如下:
public class Producer01 { //队列名称 private static final String QUEUE = "helloworld"; public static void main(String[] args) throws IOException, TimeoutException { Connection connection = null; Channel channel = null; try { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setPort(5672); factory.setUsername("guest"); factory.setPassword("guest"); factory.setVirtualHost("/");//rabbitmq默认虚拟机名称为“/”,虚拟机相当于一个独立的mq服务 器 //创建与RabbitMQ服务的TCP连接 connection = factory.newConnection(); //创建与Exchange的通道,每个连接可以创建多个通道,每个通道代表一个会话任务 channel = connection.createChannel(); /** * 声明队列,如果Rabbit中没有此队列将自动创建 * param1:队列名称 * param2:是否持久化 * param3:队列是否独占此连接 * param4:队列不再使用时是否自动删除此队列 * param5:队列参数 */ channel.queueDeclare(QUEUE, true, false, false, null); String message = "helloworld小明"+System.currentTimeMillis(); /** * 消息发布方法 * param1:Exchange的名称,如果没有指定,则使用Default Exchange * param2:routingKey,消息的路由Key,是用于Exchange(交换机)将消息转发到指定的消息队列 * param3:消息包含的属性 * param4:消息体 */ /** * 这里没有指定交换机,消息将发送给默认交换机,每个队列也会绑定那个默认的交换机,但是不能显 示绑定或解除绑定 * 默认的交换机,routingKey等于队列名称 */ channel.basicPublish("", QUEUE, null, message.getBytes()); System.out.println("Send Message is:'" + message + "'"); } catch(Exception ex) { ex.printStackTrace(); } finally { if(channel != null) { channel.close(); } if(connection != null) { connection.close(); } } } }
2.5.3.消费者
在消费者工程下的test中创建测试类如下:
public class Consumer01 { private static final String QUEUE = "helloworld"; public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); //设置MabbitMQ所在服务器的ip和端口 factory.setHost("127.0.0.1"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); //声明队列 channel.queueDeclare(QUEUE, true, false, false, null); //定义消费方法 DefaultConsumer consumer = new DefaultConsumer(channel) { /** * 消费者接收消息调用此方法 * @param consumerTag 消费者的标签,在channel.basicConsume()去指定 * @param envelope 消息包的内容,可从中获取消息id,消息routingkey,交换机,消息和重传标志 (收到消息失败后是否需要重新发送) * @param properties * @param body * @throws IOException */ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { //交换机 String exchange = envelope.getExchange(); //路由key String routingKey = envelope.getRoutingKey(); //消息id long deliveryTag = envelope.getDeliveryTag(); //消息内容 String msg = new String(body,"utf‐8"); System.out.println("receive message.." + msg); } }; /** * 监听队列String queue, boolean autoAck,Consumer callback * 参数明细 * 1、队列名称 * 2、是否自动回复,设置为true为表示消息接收到自动向mq回复接收到了,mq接收到回复会删除消息,设置 为false则需要手动回复 * 3、消费消息的方法,消费者接收到消息后调用此方法 */ channel.basicConsume(QUEUE, true, consumer); } }
2.5.4.总结
1、发送端操作流程
1)创建连接
2)创建通道
3)声明队列
4)发送消息
2、接收端
1)创建连接
2)创建通道
3)声明队列
4)监听队列
5)接收消息
6 )ack回复