rabbitmq介绍122教学讲解课件

RabbitMQ介绍

目录什么是MQMQ有什么优势哪些情况下建议使用MQ什么是RabbitMQ选择RabbitMQ理由RabbitMQ服务场景RabbitMQ结构图RabbitMQ名词解释目录RabbitMQ客户端使用流程(productor/cunsumer)Productor范例代码及注意事项Consumer范例代码及注意事项开发中注意事项及重点关注异常处理RabbitMQ服务端配置及重点参数RabbitMQ与Spring整合范例代码什么是MQ?MQ全称为MessageQueue,消息队列（MQ）是一种应用程序对应用程序的通信方法（消息传递），一般用作进程间通讯MQ有什么优势?MQ本身是异步的，往队列里发送消息后无需等待，不同于通信协议。如HTTP协议（同步），客户端发出请求后必须等待服务器回应哪些情况下建议使用MQ高并发应用来不及处理，实时性要求不高多应用之间异步通信，且耗时操作什么是RabbitMQRabbitMQ是由Erlang(爱立信公司)语言开发，实现AdvancedMessageQueuingProtocol(AMQP高级消息队列协议)的消息中间件。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。选择RabbitMQ理由Reliability可靠性Exchange交换机、Queue队列、Message消息持久化、高可用性FlexibleRouting灵活路由Clustering集群分为Disc（硬盘）与RAM（内存），保证至少一台DiscHighlyAvailableQueues高可用队列与集群结合使用，设置队列间的消息同步ManagementUI管理界面异常情况下RabbitMQ处理方式(单机丢失/网络丢失/掉电/队列爆满)单机丢失RabbitMQ支持集群，多台机器队列同步，丢失消息可从其他机器上获取网络丢失掉电RabbitMQ支持持久化，数据保存在硬盘上队列爆满RabbitMQ支持流控机制，可修改内存大小，默认为机器内存的40%RabbitMQ服务场景应用程序之间无需即时返回且耗时操作（异步）WorkQueues（消息均匀分配消息给消费者）Publish/Subscribe(广播模式，消息分发给所有的消费者)Routing(消费者接收消息由路由规则决定，简单路由名)Topics(消费者接收消息由路由规则决定，路由规则名比较复杂)RPC远程调用（同步）RabbitMQ结构图RabbitMQ名词解释Broker：消息队列服务器实体，例如RabbitMQ服务Vhost：虚拟主机，默认为“/”，一个broker里可以有多个vhost，区分不同用户权限，类似java的命令空间Connection：应用程序与broker连接，可有多个连接Channel：消息通道，connection中可建立多个channel，每个channel代表一个会话任务，所有操作都在channel中进行。RabbitMQ名词解释Exchange：消息交换机，channel中可有多个，用于投递消息。应用程序发送消息时先把消息给交换机，由交换机投递给队列，不是直接给队列Queue：队列，用于存放消息Message：消息，应用程序需要发送的数据Bind：根据routingKey绑定exchange与queue规则，决定消息发送的方向RabbitMQ对象间关系broker可多个Connection可多个Channel可多个Exchange可多个QueuemessageExchange主要3种类型Fanout：不处理路由键（没有routingKey），只需把队列绑定到交换机上。发送到交换机的消息都会转发到与该交换机绑定的所有队列上，类似于广播，转发消息是最快的Exchange主要3种类型Direct：处理路由键（有routingKey）。将一队列绑定到交换机上，该消息需与一个特定的路由键（routingKey）完全匹配Exchange主要3种类型Topic：与direct类似，功能更强，支持模糊绑定*表示通配一个词#表示通配0个或多个词RabbitMQ客户端使用流程(productor/cunsumer)Productor范例代码及注意事项//以exchange为direct为例package

com.rabbitmq.test.ow.demo2;import

java.io.IOException;import

com.rabbitmq.client.BlockedListener;import

com.rabbitmq.client.Channel;import

com.rabbitmq.client.Connection;import

com.rabbitmq.client.ConnectionFactory;import

com.rabbitmq.client.MessageProperties;import

com.rabbitmq.client.ShutdownListener;import

com.rabbitmq.client.ShutdownSignalException;import

com.rabbitmq.client.impl.AMQCommand;import

com.rabbitmq.client.impl.AMQImpl;/***

Title:

Producer.java*

Description:【生产者样例】*

@author:

zengqiang.yang*

@date:

2013-12-25**版权所有(c)2013，天翼电子商务有限公司

*/public

class

Producer{

//交换机命名规范：e_模块_其他

private

static

final

StringEXCHANGE_NAME

="e_tyb_test";

//routingkey命名规范：r_模块_其他

private

static

final

StringROUTING_KEY

="r_tyb_test";

public

static

void

main(String[]argv)throws

Exception{

//注意：factory应为单例，不要每次取消息新建一次对象

ConnectionFactoryfactory=new

ConnectionFactory();

factory.setHost("90");

factory.setPort(5672);//默认端口Productor范例代码及注意事项

factory.setUsername("guest");//默认用户名

factory.setPassword("guest");//默认密码

factory.setVirtualHost("/");//默认虚拟主机，区分权限

//设置心跳时间，防止长时间未活动被防火墙杀死,默认600秒,单位：秒//

factory.setRequestedHeartbeat(60*4);

//连接超时时间,单位：毫秒//

factory.setConnectionTimeout(1000*2);

//注意：connection应为单例，不要每次取消息新建一次对象

Connectionconnection=factory.newConnection();

//监听connection关闭异常

connection.addShutdownListener(new

ShutdownListener(){ @Override

public

void

shutdownCompleted(ShutdownSignalExceptioncause){

//connection异常

if

(cause.isHardError()){

System.out.println("connection异常:["

+cause.getMessage()+"]");

//程序引起的异常，如：connection.close()

if

(cause.isInitiatedByApplication()){

System.out.println("connection关闭异常,重连...begin...");//

connection=factory.newConnection();

System.out.println("connection关闭异常,重连...end...");

}else{//rabbitmq服务引起的异常

AMQCommandamqCommand=(AMQCommand)cause.getReason();

if(amqCommand.getMethod()instanceof

AMQImpl.Connection.Close){

AMQImpl.Connection.Closeclose=(AMQImpl.Connection.Close)amqCommand.getMethod();

if(320==close.getReplyCode()){//rabbitmq服务器强制关闭

System.out.println("connection关闭异常,请检查rabbitmq服务器是否正常启动！");

}}}}}});Productor范例代码及注意事项

//监听connection阻塞异常

connection.addBlockedListener(new

BlockedListener(){

@Override

public

void

handleUnblocked()throws

IOException{

System.out.println("connection已解除阻塞！");

}

@Override

public

void

handleBlocked(Stringreason)throws

IOException{

System.out.println("connection阻塞原因：["+reason+"],请检查内存是否够！");

}

});

//注意：channel应为单例，不要每次取消息新建一次对象

Channelchannel=connection.createChannel();

//监听channel关闭异常

channel.addShutdownListener(new

ShutdownListener(){

@Override

public

void

shutdownCompleted(ShutdownSignalExceptioncause){

//channel异常

if

(!cause.isHardError()){System.out.println("channel异常:["

+cause.getMessage()+"]");

}

}

});

Productor范例代码及注意事项

/** *创建交换机 *exchange 交换机名 *type

交换机类型 fanout:广播模式,所有消费者都能收到生产者发送的消息,速度更快,不需设置routingkey * direct:只有与routingkey配置的消费者才能收到消息 * topic:与direct相同,只是支持模糊配置,类似正则表达式,功能更强, **表示通配一个词,#表示通配0个或多个词（注意：是词,不是字母） * *durable

durable=true,交换机持久化,rabbitmq服务重启交换机依然存在,保证不丢失; * durable=false,相反 * */

channel.exchangeDeclare(EXCHANGE_NAME,"direct",true);

//模拟发送消息

for

(int

i=0;i<100;i++){

Stringmessage="HelloWorld!";

message+=i;

/***exchange 交换机名,""为默认交换机，direct类型

*routingKey

exchange为direct、topic类型时指定routingKey,exchange为fanout类型时指定queueName队列名 *props

MessageProperties.PERSISTENT_TEXT_PLAIN:消息持久化,rabbitmq服务重启消息不会丢失;null:非持久化 *body 发送消息 */

channel.basicPublish(EXCHANGE_NAME,ROUTING_KEY,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());

System.out.println("[x]Sent'"

+message+"'");

}

//关闭连接//

channel.close();//

connection.close();

}}Productor范例代码及注意事项ConnectionFactory、Connection、Channel注意单例控制默认心跳时间为10分钟，发现个人账户环境防火墙为5分钟，有可能被防火墙杀死，建议设置4分钟connectionFactory.setRequestedHeartbeat(4*60);添加Shutdown、Blocked异常监听，Shutdown后重连机制、Blocked后的日志输出Consumer范例代码及注意事项//以exchange为direct为例package

com.rabbitmq.test.ow.demo2;import

java.io.IOException;import

java.util.HashMap;import

java.util.Map;import

com.rabbitmq.client.BlockedListener;import

com.rabbitmq.client.Channel;import

com.rabbitmq.client.Connection;import

com.rabbitmq.client.ConnectionFactory;import

com.rabbitmq.client.QueueingConsumer;import

com.rabbitmq.client.ShutdownListener;import

com.rabbitmq.client.ShutdownSignalException;import

com.rabbitmq.client.impl.AMQCommand;import

com.rabbitmq.client.impl.AMQImpl;/***

Title:

Consumer.java*

Description:【消费者样例】*

@author:

zengqiang.yang*

@date:

2013-12-25**版权所有(c)2013，天翼电子商务有限公司

*/Consumer范例代码及注意事项public

class

Consumer{

//交换机命名规范：e_模块_其他

private

static

final

StringEXCHANGE_NAME

="e_tyb_test";

//队列命名规范：q_模块_其他

private

static

final

StringQUEUE_NAME

="q_tyb_test";

//routingkey命名规范：r_模块_其他

private

static

final

StringROUTING_KEY

="r_tyb_test";

public

static

void

main(String[]argv)throws

Exception{

//注意：factory应为单例，不要每次取消息新建一次对象

ConnectionFactoryfactory=new

ConnectionFactory();

factory.setHost("90");

factory.setPort(5672);//默认端口

factory.setUsername("guest");//默认用户名

factory.setPassword("guest");//默认密码

factory.setVirtualHost("/");//默认虚拟主机，区分权限

//设置心跳时间，防止长时间未活动被防火墙杀死,默认600秒,单位：秒//

factory.setRequestedHeartbeat(60*4);Consumer范例代码及注意事项

//连接超时时间,单位：毫秒//

factory.setConnectionTimeout(1000*2);

//注意：connection应为单例，不要每次取消息新建一次对象

Connectionconnection=factory.newConnection();

//监听connection关闭异常

connection.addShutdownListener(new

ShutdownListener(){

@Override

public

void

shutdownCompleted(ShutdownSignalExceptioncause){

//connection异常

if

(cause.isHardError()){

System.out.println("connection异常:["

+cause.getMessage()+"]");

//程序引起的异常，如：connection.close()

if

(cause.isInitiatedByApplication()){

System.out.println("connection关闭异常,重连...begin...");//

connection=factory.newConnection();

System.out.println("connection关闭异常,重连...end...");

}

Consumer范例代码及注意事项else{//rabbitmq服务引起的异常

AMQCommandamqCommand=(AMQCommand)cause.getReason();

if(amqCommand.getMethod()instanceof

AMQImpl.Connection.Close){

AMQImpl.Connection.Closeclose=(AMQImpl.Connection.Close)amqCommand.getMethod();

if(320==close.getReplyCode()){//rabbitmq服务器强制关闭

System.out.println("connection关闭异常,请检查rabbitmq服务器是否正常启动！");

}

}

}

}

}

});

//监听connection阻塞异常

connection.addBlockedListener(new

BlockedListener(){

@Override

public

void

handleUnblocked()throws

IOException{

System.out.println("connection已解除阻塞！");

}

@Override

public

void

handleBlocked(Stringreason)throws

IOException{

System.out.println("connection阻塞原因：["+reason+"],请检查内存是否够！");

}

});Consumer范例代码及注意事项

//注意：channel应为单例，不要每次取消息新建一次对象

Channelchannel=connection.createChannel();

//监听channel关闭异常

channel.addShutdownListener(new

ShutdownListener(){

@Override

public

void

shutdownCompleted(ShutdownSignalExceptioncause){

//channel异常

if

(!cause.isHardError()){

System.out.println("channel异常:["

+cause.getMessage()+"]");

}

}

});

//同一时间一个消费者只能接收一条消息,web管理界面队列Unacked数值，如多个队列数值累加

channel.basicQos(1);

Consumer范例代码及注意事项/***创建交换机*exchange 交换机名

*type 交换机类型 fanout:广播模式,所有消费者都能收到生产者发送的消息,速度更快,不需设置routingkey* direct:只有与routingkey配置的消费者才能收到消息* topic:与direct相同,只是支持模糊配置,类似正则表达式,功能更强,**表示通配一个词,#表示通配0个或多个词（注意：是词,不是字母）* *durable

durable=true,交换机持久化,rabbitmq服务重启交换机依然存在,保证不丢失;* durable=false,相反**/

channel.exchangeDeclare(EXCHANGE_NAME,"direct",true);

/***创建队列*queue 队列名*durable 持久化队列，true:服务器重启队列不会丢失;false:反之*exclusive 排他性，true:首次申明的connection连接下可见;* false：所有connection连接下都可见*connection关闭后队列自动删除,忽略队列持久化*autoDelete

true:无消费者时，队列自动删除；false：无消费者时，队列不会自动删除*arguments 可指定队列里消息总数*/

Map<String,Object>args

=new

HashMap<String,Object>();//

args.put("x-max-length",100);

//设置队列里的最大消息数//

args.put("x-message-ttl",1000*10);

//设置消息过期时间,时间一过消息自动删除,单位：毫秒//

channel.queueDeclare(QUEUE_NAME,true,false,false,args);

channel.queueDeclare(QUEUE_NAME,true,false,false,null);Consumer范例代码及注意事项

channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,ROUTING_KEY);

System.out.println("[*]Waitingformessages.ToexitpressCTRL+C");

//声明一个消费者

QueueingConsumerconsumer=new

QueueingConsumer(channel);

/** *queue 队列名 *autoAck

true:消息从队列删除，不管是否正确处理;false：消息不从队列删除，需要ack响应 *callback 消费者 */

channel.basicConsume(QUEUE_NAME,false,consumer);

while

(true){

//循环获取消息

QueueingConsumer.Deliverydelivery=consumer.nextDelivery();

Stringmessage=new

String(delivery.getBody());

System.out.println("[x]Received'"

+message+"'");

doWork(message);

System.out.println("[x]Done");

//确认，消息从队列中删除

channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);

//拒绝，消息不从队列删除//

channel.basicReject(delivery.getEnvelope().getDeliveryTag(),true);

}

}

private

static

void

doWork(Stringtask)throws

InterruptedException{

for

(char

ch:task.toCharArray()){

if

(ch==''){

Thread.sleep(1000);

}

}

}}Consumer范例代码及注意事项ConnectionFactory、Connection、Channel注意单例控制默认心跳时间为10分钟，发现个人账户环境防火墙为5分钟，有可能被防火墙杀死，建议设置4分钟connectionFactory.setRequestedHeartbeat(4*60);添加Shutdown、Blocked异常监听，Shutdown后重连机制、Blocked后的日志输出对于比较重要的消息，消费者启动应在生产者之前，避免生产者发送消息时消费者未启动消息丢失开发中注意事项及重点关注异常处理ConnectionFactory、Connection、Channel切勿每发送一条消息创建一次，因用单例控制Connection默认心跳 //默认600秒（10分钟）connectionFactory.setRequestedHeartbeat(240);Exchange持久化 //durable=true持久化，durable=false非持久化 channel.exchangeDeclare(EXCHANGE_NAME,“direct”,true);开发中注意事项及重点关注异常处理Queue持久化 //durable=true持久化，durable=false非持久化 channel.queueDeclare(QUEUE_NAME,true,false,false,null);Message持久化 //props=MessageProperties.PERSISTENT_TEXT_PLAIN消息持久化 channel.basicPublish("",QUEUE_NAME,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());开发中注意事项及重点关注异常处理Queue队列排他性 //exclusive=true首次申明的connection连接下可见，

//exclusive=false所有connection连接下都可见 channel.queueDeclare(QUEUE_NAME,true,true,false,null);Queue队列里消息数大小设置 Map<String,Object>args=newHashMap<String,Object>(); args.put("x-max-length",50);//设置队列里的最大消息数 channel.queueDeclare(QUEUE_NAME,true,false,false,args);开发中注意事项及重点关注异常处理Queue队列里消息有效性 Map<String,Object>args=newHashMap<String,Object>(); //设置消息过期时间,时间一过消息自动删除,单位：毫秒 args.put("x-message-ttl",1000*10); channel.queueDeclare(QUEUE_NAME,true,false,false,args);开发中注意事项及重点关注异常处理关闭异常ShutdownSignalException通过Connection.addShutdownListener监听，可分为程序引起（如：connection.close()）和服务器引起两大异常，程序引起建议捕获异常后有重连机制阻塞异常通过connection.addBlockedListener监听，当使用内存接近最大内存时，消息会阻塞（handleBlocked方法

捕获），可通过调整内存大小解除阻塞（handleUnblocked方法捕获）RabbitMQ服务端配置及重点参数添加用户并指定密码./rabbitmqctladd_user用户名

密码设置用户权限./rabbitmqctlset_user_tags用户名

权限组权限组4类Management PolicymakerMonitoringAdministratorManagement查看rabbitmq服务总消息数、当前与多少机器相连（connection）、建了多少通道（channel）、建立了多少队列数等Policymaker比management多了一个Admin选项卡中的policies标签，可添加镜像队列，满足集群情况下队列的同步Monitoring比management多了Overview选项卡中的Nodes、Portsandcontexts标签，可查看当前打开socket连接数、rabbitmq内存、硬盘使用情况；amqp端口（默认5672）及Webmanagement管理界面端口（默认15672，55672是rabbitmq3.0之前版本端口号）Administrator最高权限，用户、虚拟主机、策略等都可设置RabbitMQ服务端配置及重点参数添加vhost./rabbitmqctladd_vhostvhost名设置vhost权限./rabbitmqctlset_permissions-pvhost名

用户名".*"".*"".*“Vhost下的所有资源都有读写权限RabbitMQ服务端配置及重点参数添加镜像队列策略./rabbitmqctl-n节点名set_policy-pvhost名

策略名"^q\_tyb\_"'{"ha-mode":"all","ha-sync-mode":"automatic"}‘^q\_tyb\_设置队列名规则ha-mode:all所有机器都接收（集群环境下）ha-sync-mode:automatic某一台机器停止重启后，自动同步队列里的消息RabbitMQ服务端配置及重点参数最大打开文件数及socket连接数如ConnectionFactory、Connection、Channel未控制单例，每发送一条消息就创建一次连接，file、socket数也随之增加，file与socket总数比例大约80%RabbitMQ服务端配置及重点参数rabbitmq内存、硬盘使用空间rabbitmq最大内存默认为机器内存的40%rabbitmq磁盘空间默认为rabbitmq数据所在分区的可用空闲空间RabbitMQ服务端配置及重点参数Exchange交换机的持久化交换机持久化,rabbitmq服务重启exchange依然存在,数据保存在硬盘上RabbitMQ服务端配置及重点参数Queue队列的持久化队列持久化,rabbitmq服务重启queue依然存在,数据保存在硬盘上

RabbitMQ服务端配置及重点参数队列最大消息数（可选）当队列里消息数到达最大值时，后面消息会替换原先的消息，队列里消息是先进先出原则代码Map<String,Object>args=newHashMap<String,Object>(); args.put("x-max-length",50);//设置队列里的最大消息数 channel.queueDeclare(QUEUE_NAME,true,false,false,args);示意图RabbitMQ服务端配置及重点参数消息有效性（可选）消息进入队列过了指定时间（单位：毫秒），自动删除消息代码 Map<String,Object>args=newHashMap<String,Object>(); //设置消息过期时间,时间一过消息自动删除,单位：毫秒 args.put("x-message-ttl",1000*10); channel.queueDeclare(QUEUE_NAME,true,false,false,args);示意图RabbitMQ服务端配置及重点参数多机集群队列同步如果一台RabbitMQ服务挂了，其他机器上会有一份副本。RabbitMQ服务端配置及重点参数队列命名规范，如：q_模块_其他影响队列同步，多机集群间队列的同步需要使用policy策略来保证RabbitMQ与Spring整合范例代码生产者applicationContext-rabbitmq-producer.xml配置文件<?xml

version="1.0"

encoding="UTF-8"?><beans

xmlns=""

xmlns:xsi=""xmlns:context=""

xmlns:rabbit=""

xsi:schemaLocation="

">

<!--连接服务配置-->

<rabbit:connection-factory

id="connectionFactory"

host="90"

username="guest"

password="guest"

port="5672"

/>

<rabbit:admin

connection-factory="connectionFactory"

/>

RabbitMQ与Spring整合范例代码<!--queue队列声明-->

<!--durable=true,交换机持久化,rabbitmq服务重启交换机依然存在,保证不丢失;durable=false,相反-->

<!--auto-delete=true:无消费者时，队列自动删除;auto-delete=false：无消费者时，队列不会自动删除-->

<!--排他性，exclusive=true:首次申明的connection连接下可见;exclusive=false：所有connection连接下都可见-->

<rabbit:queue

id="q_tyb_test2"

durable="true"

auto-delete="false"

exclusive="false"

name="q_tyb_test2"

/>

<!--exchangequeuebingingkey绑定-->

<!--durable=true,交换机持久化,rabbitmq服务重启交换机依然存在,保证不丢失;durable=false,相反-->

<!--auto-delete=true:无消费者时，队列自动删除;auto-delete=false：无消费者时，队列不会自动删除-->

<rabbit:direct-exchangename="e_tyb_test2"

durable="true"

auto-delete="false"

id="e_tyb_test2">

<rabbit:bindings>

<rabbit:binding

queue="q_tyb_test2"

key="r_tyb_test2"

/>

</rabbit:bindings>

</rabbit:direct-exchange>

<rabbit:template

id="amqpTemplate"

exchange="e_tyb_test2"

connection-factory="connectionFactory"

/></beans>RabbitMQ与Spring整合范例代码//生产者packagecom.rabbitmq.test.ow.spring;importorg.springframework.amqp.core.AmqpTemplate;importorg.springframework.context.ApplicationContext;importorg.springframework.context.support.;/***

Title:

Producer.java*

Description:【rabbitmq与spring整合-生产者】*

@author:

zengqiang.yang*

@date:

2013-12-30**版权所有(c)2013，天翼电子商务有限公司

*/public

classProducer{public

static

voidmain(String[]args){ApplicationContextcontext=new

("src/main/resources/applicationContext-rabbitmq-producer.xml");RabbitMQ与Spring整合范例代码AmqpTemplateamqpTemplate=(AmqpTemplate)context.getBean("amqpTemplate");//模拟发送消息for(inti=0;i<100;i++){Stringmessage="HelloWorld!";message+=i;/***r_tyb_test2routingkey*message发送消息*/amqpTemplate.convertAndSend("r_tyb_test2",message);System.out.println("Sent:"+message);}}}RabbitMQ与Spring整合范例代码消费者applicationContext-rabbitmq-consumer.xml配置文件<?xml

version="1.0"

encoding="UTF-8"?><beans

xmlns=""

xmlns:xsi=""

xmlns:context=""

xmlns:rabbit=""

xsi:schemaLocation="

">

<!--连接服务配置-->

<rabbit:connection-factory

id="connectionFactory"

host="90"

username="guest"

password="guest"

port="5672"

/>

<rabbit:admin

connection-factory="connectionFactory"

/>

<!--queue队列声明-->

<!--durable=true,交换机持久化,rabbitmq服务重启交换机依然存在,保证不丢失;durable=false,相反-->

<!--auto-delete=true:无消费者时，队列自动删除;auto-delete=false：无消费者时，队列不会自动删除-->

<!--排他性，exclusive=true:首次申明的connection连接下可见;exclusive=false：所有connection