欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

RabbitMQ的简单模式快速入门与超时异常的处理方法

程序员文章站 2022-06-13 08:12:03
本文适合JAVA新人,想了解RabbitMQ又不想去看官网文档的人(英语水看的头疼(◎﹏◎),但建议有能力还是去看官网文档)。 消息队列MQ(一) MQ全称为Message Queue,消息队列是应用程序和应用程序之间的通信方法。 先引入一下常见的通讯方案。 为什么使用MQ? 在项目中,可将一些无需 ......

本文适合java新人,想了解rabbitmq又不想去看官网文档的人(英语水看的头疼(◎﹏◎),但建议有能力还是去看官网文档)。

消息队列mq(一)

mq全称为message queue,消息队列是应用程序和应用程序之间的通信方法。

先引入一下常见的通讯方案。

RabbitMQ的简单模式快速入门与超时异常的处理方法

为什么使用mq

在项目中,可将一些无需即时返回且耗时的操作提取出来,进行异步处理,而这种异步处理的方式大大的节省服务器的请求响应时间,从而提高了系统的吞吐量

 开发中消息队列通常有如下应用场景:

应用解耦、异步处理(提高系统响应速度)、流量削峰(高峰堆积消息,峰后继续处理消息)、日志处理(分布式日志,一般使用kafka)、纯粹通讯。

 RabbitMQ的简单模式快速入门与超时异常的处理方法

amqp 和 jms

mq是消息通信的模型;实现mq的大致有两种主流方式:amqp、jms。

amqp

amqp高级消息队列协议,是一个进程间传递异步消息的网络协议,更准确的说是一种binary wire-level protocol(链接协议)。这是其和jms的本质差别,amqp不从api层进行限定,而是直接定义网络交换的数据格式。

jms

jms即java消息服务(javamessage service)应用程序接口,是一个java平台中关于面向消息中间件(mom)的api,用于在两个应用程序之间,或分布式系统中发送消息,进行异步通信。

 amqp 与 jms 区别

jms是定义了统一的接口,来对消息操作进行统一;amqp是通过规定协议来统一数据交互的格式

jms限定了必须使用java语言;amqp只是协议,不规定实现方式,因此是跨语言的。

jms规定了两种消息模式;而amqp的消息模式更加丰富。

消息队列产品:目前市面上成熟主流的mq有kafka 、rocketmq、rabbitmq,本文主要介绍rabbitmq使用。

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

rabbitmq介绍

rabbitmq是由erlang语言开发,基于amqp(advanced message queue 高级消息队列协议)协议实现的消息队列,它是一种应用程序之间的通信方法,消息队列在分布式系统开发中应用非常广泛。

rabbitmq官方地址:

rabbitmq提供了6种模式:简单模式,work工作队列(集群)模式,publish/subscribe发布与订阅(交换机的广播)模式,routing(交换机的定向)路由模式,topics主题(路由灵活)模式,rpc远程调用模式(远程调用,不太算mq;不作介绍);//括号内的是自己的理解方式仅供参考。详细可以去看官方介绍。

官网对应模式介绍:

安装rabbirmq

两种方式:windows环境与linux环境(这里跳过)

我是linuxcenos6.7安装的3.6.10版本

启动成功参考如下两张图

RabbitMQ的简单模式快速入门与超时异常的处理方法

RabbitMQ的简单模式快速入门与超时异常的处理方法

先在web页面管理用户

RabbitMQ的简单模式快速入门与超时异常的处理方法

角色说明: tags

1、超级管理员(administrator)

可登陆管理控制台,可查看所有的信息,并且可以对用户,策略(policy)进行操作。

2、监控者(monitoring)

可登陆管理控制台,同时可以查看rabbitmq节点的相关信息(进程数,内存使用情况,磁盘使用情况等)

3、策略制定者(policymaker) :可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分)。

4、普通管理者(management):仅可登陆管理控制台,无法看到节点信息,也无法对策略进行管理。

5、其他 :  无法登陆管理控制台,通常就是普通的生产者和消费者。

virtual hosts配置

在rabbitmq中可以虚拟消息服务器virtual host,每个virtual hosts相当于一个相对独立的rabbitmq服务器,每个virtualhost之间是相互隔离的。exchange、queue、message不能互通。 相当于mysql的db。virtual name一般以/开头。

添加队列,这里需要将上下两张图结合起来看

RabbitMQ的简单模式快速入门与超时异常的处理方法

RabbitMQ的简单模式快速入门与超时异常的处理方法

需改用户的密码

RabbitMQ的简单模式快速入门与超时异常的处理方法

查看默认的交换机

RabbitMQ的简单模式快速入门与超时异常的处理方法

常见的端口

RabbitMQ的简单模式快速入门与超时异常的处理方法

rabbitmq入门

目标:入门案例将使用rabbitmq的简单模式实现通讯过程。

RabbitMQ的简单模式快速入门与超时异常的处理方法

1.创建maven工程,先在pom.xml添加依赖。

 1 <?xml version="1.0" encoding="utf-8"?>
 2 <project xmlns="http://maven.apache.org/pom/4.0.0"
 3          xmlns:xsi="http://www.w3.org/2001/xmlschema-instance"
 4          xsi:schemalocation="http://maven.apache.org/pom/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 5     <modelversion>4.0.0</modelversion>
 6     <groupid>com.jxjdemo</groupid>
 7     <artifactid>rabbitmq1_demo</artifactid>
 8     <version>1.0-snapshot</version>
 9 
10     <dependencies>
11         <dependency> <!--rabbitmq的依赖-->
12             <groupid>com.rabbitmq</groupid>
13             <artifactid>amqp-client</artifactid>
14             <version>5.6.0</version>
15         </dependency>
16     </dependencies>
17 </project>

2.新建生产者类,生产发送消息

RabbitMQ的简单模式快速入门与超时异常的处理方法

 1 package com.jxjdemo.mq.simple;
 2 
 3 import com.rabbitmq.client.channel;
 4 import com.rabbitmq.client.connection;
 5 import com.rabbitmq.client.connectionfactory;
 6 
 7 public class simpleproducer {
 8     public static void main(string args[]) throws exception{
 9 //1、创建链接工厂对象-factory=newconnectionfactory()。创建链接用
10         connectionfactory factory = new connectionfactory();
11 
12 //2、设置rabbitmq服务主机地址,默认localhost-factory.sethost("localhost")
13         factory.sethost("192.168.211.128");
14 //3、设置rabbitmq服务端口,默认-1-factory.setport(5672)
15         factory.setport(5672);
16 //4、设置虚拟主机名字,默认/-factory.setvirtualhost("szitheima")
17          factory.setvirtualhost("shujuku1122");
18 //5、设置用户连接名,默认guest-factory.setusername("admin")
19         factory.setusername("admin");
20 //6、设置链接密码,默认guest-factory.setpassword("admin")
21         factory.setpassword("123456");
22  //      factory.setconnectiontimeout(5000);
23 //        factory.setworkpooltimeout(5000);
24 //        factory.sethandshaketimeout(5000);
25 //7、创建链接-connection=factory.newconnection()
26         connection connection = factory.newconnection(); //报错,抛异常
27 //8、创建频道-channel=connection.createchannel()
28         channel channel = connection.createchannel();
29 //9、声明队列-channel.queuedeclare(名称,是否持久化(true先存硬盘,读完再删),是否独占本连接,是否自动删除(false读完再删),附加参数)
30         channel.queuedeclare("simplequeue", true, false, false, null);
31 //10、创建消息-stringm=xxx
32         string msg = "这是我们第一次发送 mq消息";
33 //11、消息发送-channel.basicpublish(交换机[默认defaultexchage],路由key[简单模式可以传递队列名称],消息其它属性,消息内容)
34         channel.basicpublish("", "simplequeue", null, msg.getbytes("utf-8"));
35 //12、关闭资源-channel.close();connection.close()
36         channel.close();
37        connection.close();
38     }
39 }

执行后发个消息,没看到异常。

RabbitMQ的简单模式快速入门与超时异常的处理方法

扩展:这里遇到的异常有,时间超时

解决方法一:

发送不成功报错,就先重启mq,在重启【管理员的方式启动】ide,一般都是mq的问题。

发送消息为空,消息不能有空格。注意库名字。

解决方法二:

我们安装系统会给系统起个名字导致:修改后的主机名并没有在linux系统的hosts文件中,因此解析的时候,无法直接从该文件中获取,需要多重解析,才能解析该主机名。

 不同的linux版本,这个配置文件也可能不同vim /etc/hosts

RabbitMQ的简单模式快速入门与超时异常的处理方法

 继续说发送成功的事情。

RabbitMQ的简单模式快速入门与超时异常的处理方法

3.创建消费者,接收消息。

RabbitMQ的简单模式快速入门与超时异常的处理方法 1 package com.jxjdemo.mq.simple;

 2 
 3 import com.rabbitmq.client.*;
 4 
 5 import javax.security.auth.callback.callback;
 6 import java.io.ioexception;
 7 import java.util.concurrent.timeoutexception;
 8 
//这里删除了文档注释
16 public class simpleconsumer { 17 public static void main(string args[]) throws ioexception, timeoutexception { 18 //1、创建链接工厂对象-factory=newconnectionfactory() 19 connectionfactory factory = new connectionfactory(); 20 //2、设置rabbitmq服务主机地址,默认localhost-factory.sethost("localhost") 21 factory.sethost("192.168.211.128"); 22 //3、设置rabbitmq服务端口,默认-1-factory.setport(5672) 23 factory.setport(5672); 24 //4、设置虚拟主机名字,默认/-factory.setvirtualhost("szitheima") 25 factory.setvirtualhost("shujuku1122"); 26 //5、设置用户连接名,默认guest-factory.setusername("admin") 27 factory.setusername("admin"); 28 //6、设置链接密码,默认guest-factory.setpassword("admin") 29 factory.setpassword("123456"); 30 //7、创建链接-connection=factory.newconnection() 31 connection connection = factory.newconnection(); 32 //8、创建频道-channel=connection.createchannel() 33 channel channel = connection.createchannel(); 34 //9、声明队列-channel.queuedeclare(名称,是否持久化,是否独占本连接,是否自动删除,附加参数) 35 channel.queuedeclare("simplequeue",true ,false , false,null ); 36 //10接收消息 37 consumer callback = new defaultconsumer(channel){ 38 /** 39 * @param consumertag 消费者标签,在channel.basicconsume时候可以指定 40 * @param envelope 信封,消息包的内容,可从中获取消息id,消息routingkey,交换机,消息和重传标志(收到消息失败后是否需要重新发送) 41 * @param properties 属性信息(生产者的发送时指定) 42 * @param body 消息内容 43 * @throws ioexception 44 */ 45 @override 46 public void handledelivery(string consumertag, envelope envelope, amqp.basicproperties properties, byte[] body) throws ioexception { 47 long deliverytag = envelope.getdeliverytag(); //消息id 48 string exchange = envelope.getexchange(); 49 string routingkey = envelope.getroutingkey(); //路由key 50 //消息内容 51 string msg = new string(body,"utf-8"); 52 system.out.println( 53 "routingkey:" + routingkey + 54 "routingkey:" + routingkey + 55 ",exchange:" + exchange + 56 ",deliverytag:" + deliverytag + 57 ",message:" + msg); 58 } 59 }; 60 channel.basicconsume("simplequeue", callback); 61 //不关闭,继续接受消息 62 } 63 }

执行后看到一下结果

RabbitMQ的简单模式快速入门与超时异常的处理方法

当你的代码运行到这里,那么恭喜你入门成功。

这次暂时先到这里结束。欲知其他4种模式且看下回慢慢分解。