rabbitMQ的api及规范说明书

rabbitMQapi与规范说明书1简介1.1背景各系统间有着很多交互，为了使系统间的相互调用更优雅，对系统无侵入，不对现有结构造成冲击，需要引入连接各系统的介质：消息中间件。1.2术语定义序号简称/术语说明1Mq一种应用程序对应用程序的通信方法.AMQP一个提供统一消息服务的应用层标准高级消息队列协议ErlangErlang是一种通用的面向并发的编程语言，目的是创造一种可以应对大规模并发活动的编程语言和运行环境。2rabbitMQ2.1rabbitMQ简介2.1.1概述RabbitMQ是基于AMQP的一款消息管理系统,RabbitMQ基于Erlang语言开发.官网：/官方教程：/getstarted.html2.1.2流程图2.1.2发消息流程图2.1.3rabbitMQ的图形化界面添加用户创建VirtualHosts给某用户设置权限查看权限界面详解2.1.4rabbitMQ的工作模式简单消息模型简单消息模型图:生产者: 连接工具类:

publicclassConnectionUtil{/**publicclassConnectionUtil{/***建立与RabbitMQ的连接*@return*@throwsException*/publicstaticConnectiongetConnection()throwsException{//定义连接工厂ConnectionFactoryfactory=newConnectionFactory();//设置服务地址factory.setHost("");//端口factory.setPort(5672);//设置账号信息，用户名、密码、vhostfactory.setVirtualHost("/abc");factory.setUsername("guest");factory.setPassword("guest");//通过工程获取连接Connectionconnection=factory.newConnection();returnconnection;}}。]

消费者获取消息:publicclassSend{privatefinalstaticStringQUEUE_NAME="simple_queue";publicclassSend{privatefinalstaticStringQUEUE_NAME="simple_queue";publicstaticvoidmain(String[]argv)throwsException{//获取到连接Connectionconnection=ConnectionUtil.getConnection();//从连接中创建通道，使用通道才能完成消息相关的操作Channelchannel=connection.createChannel();//声明（创建）队列channel.queueDeclare(QUEUE_NAME,false,false,false,null);//消息内容Stringmessage="HelloWorld!";//向指定的队列中发送消息channel.basicPublish("",QUEUE_NAME,null,message.getBytes());System.out.println("[x]Sent'"+message+"'");//关闭通道和连接channel.close();connection.close();}}

publicclassRecv{privatefinalstaticStringQUEUE_NAME="simple_queue";publicclassRecv{privatefinalstaticStringQUEUE_NAME="simple_queue";publicstaticvoidmain(String[]argv)throwsException{//获取到连接Connectionconnection=ConnectionUtil.getConnection();//创建通道Channelchannel=connection.createChannel();//声明队列channel.queueDeclare(QUEUE_NAME,false,false,false,null);//定义队列的消费者DefaultConsumerconsumer=newDefaultConsumer(channel){//获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用@OverridepublicvoidhandleDelivery(StringconsumerTag,Envelopeenvelope,BasicPropertiesproperties,byte[]body)throwsIOException{//body即消息体Stringmsg=newString(body);System.out.println("[x]received:"+msg+"!");}};//监听队列，第二个参数：是否自动进行消息确认。channel.basicConsume(QUEUE_NAME,true,consumer);}}work消息模型work消息模型图:角色:P:生产者任务的发布者C1：消费者，领取任务并且完成任务C2：消费者2：领取任务并完成任务publicclassSend{privatefinalstaticStringQUEUE_NAME="test_work_queue";publicclassSend{privatefinalstaticStringQUEUE_NAME="test_work_queue";publicstaticvoidmain(String[]argv)throwsException{//获取到连接Connectionconnection=ConnectionUtil.getConnection();//获取通道Channelchannel=connection.createChannel();//声明队列channel.queueDeclare(QUEUE_NAME,false,false,false,null);消费者://循环发布任务for(inti=0;i<50;i++){//循环发布任务for(inti=0;i<50;i++){//消息内容Stringmessage="task.."+i;channel.basicPublish("",QUEUE_NAME,null,message.getBytes());System.out.println("[x]Sent'"+message+"'");Thread.sleep(i*2);}//关闭通道和连接channel.close();connection.close();}}订阅模型-Fanout订阅模型示意图:角色:在广播模式下，消息发送流程是这样的：1）可以有多个消费者2）每个消费者有自己的queue（队列）3）每个队列都要绑定到Exchange（交换机）4）生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定。5）交换机把消息发送给绑定过的所有队列6）队列的消费者都能拿到消息。实现一条消息被多个消费者消费生产者:两个变化：1）声明Exchange，不再声明QueuepublicclassSend{privatefinalstaticStringEXCHANGE_NAME="fanout_exchange_test";publicclassSend{privatefinalstaticStringEXCHANGE_NAME="fanout_exchange_test";publicstaticvoidmain(String[]argv)throwsException{//获取到连接Connectionconnection=ConnectionUtil.getConnection();//获取通道Channelchannel=connection.createChannel();//声明exchange，指定类型为fanoutchannel.exchangeDeclare(EXCHANGE_NAME,"fanout");//消息内容Stringmessage="Helloeveryone";//发布消息到Exchangechannel.basicPublish(EXCHANGE_NAME,"",null,message.getBytes());System.out.println("[生产者]Sent'"+message+"'");channel.close();connection.close();}}publicclassRecv{publicclassRecv{privatefinalstaticStringQUEUE_NAME="fanout_exchange_queue_1";privatefinalstaticStringEXCHANGE_NAME="fanout_exchange_test";publicstaticvoidmain(String[]argv)throwsException{//获取到连接Connectionconnection=ConnectionUtil.getConnection();//获取通道Channelchannel=connection.createChannel();//声明队列channel.queueDeclare(QUEUE_NAME,false,false,false,null);//绑定队列到交换机channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"");//定义队列的消费者DefaultConsumerconsumer=newDefaultConsumer(channel){//获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用@OverridePublicvoidhandleDelivery(StringconsumerTag,Envelopeenvelope,BasicPropertiesproperties,byte[]body)throwsIOException{//body即消息体Stringmsg=newString(body);System.out.println("[消费者1]received:"+msg+"!");}};//监听队列，自动返回完成channel.basicConsume(QUEUE_NAME,true,consumer);}}要注意代码中：队列需要和交换机绑定消费者2: 与消费者1除了需要声明的QUEUE_NAME不一致之外,其他都一致订阅模型-Direct订阅模型示意图:图解：P：生产者，向Exchange发送消息，发送消息时，会指定一个routingkey。X：Exchange（交换机），接收生产者的消息，然后把消息递交给与routingkey完全匹配的队列C1：消费者，其所在队列指定了需要routingkey为error的消息publicclassSend{privatefinalstaticStringEXCHANGE_NAME="direct_exchange_test";publicclassSend{privatefinalstaticStringEXCHANGE_NAME="direct_exchange_test";publicstaticvoidmain(String[]argv)throwsException{//获取到连接Connectionconnection=ConnectionUtil.getConnection();生产者:publicclassRecv{privatefinalstaticStringQUEUE_NAME="direct_exchange_queue_1";privatefinalstaticStringEXCHANGE_NAME="direct_exchange_test";publicstaticvoidmain(String[]argv)throwsException{publicclassRecv{privatefinalstaticStringQUEUE_NAME="direct_exchange_queue_1";privatefinalstaticStringEXCHANGE_NAME="direct_exchange_test";publicstaticvoidmain(String[]argv)throwsException{//获取到连接Connectionconnection=ConnectionUtil.getConnection();//获取通道Channelchannel=connection.createChannel();//声明队列channel.queueDeclare(QUEUE_NAME,false,false,false,null);//获取通道Channelchannel=connection.createChannel();//声明exchange，指定类型为directchannel.exchangeDeclare(EXCHANGE_NAME,"direct");//消息内容Stringmessage="商品新增了，id=1001";//发送消息，并且指定routingkey为：aaachannel.basicPublish(EXCHANGE_NAME,"aaa",null,message.getBytes());System.out.println("[商品服务：]Sent'"+message+"'");channel.close();connection.close();}}//绑定队列到交换机，同时指定需要订阅的routingkey。//绑定队列到交换机，同时指定需要订阅的routingkey。channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"aaa");channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"bbb");//定义队列的消费者DefaultConsumerconsumer=newDefaultConsumer(channel){//获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用@OverridepublicvoidhandleDelivery(StringconsumerTag,Envelopeenvelope,BasicPropertiesproperties,byte[]body)throwsIOException{//body即消息体Stringmsg=newString(body);System.out.println("[消费者1]received:"+msg+"!");}};//监听队列，自动ACKchannel.basicConsume(QUEUE_NAME,true,consumer);}}而且可以有多个消费者订阅模型-Topic说明:Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routingkey的时候使用通配符！Routingkey一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如：item.insert图示:解释:红色Queue：绑定的是usa.#，因此凡是以usa.开头的routingkey都会被匹配到黄色Queue：绑定的是#.news，因此凡是以.news结尾的routingkey都会被匹配publicclassSend{privatefinalstaticStringEXCHANGE_NAME="topic_exchange_test";publicclassSend{privatefinalstaticStringEXCHANGE_NAME="topic_exchange_test";publicstaticvoidmain(String[]argv)throwsException{//获取到连接Connectionconnection=ConnectionUtil.getConnection();//获取通道Channelchannel=connection.createChannel();//声明exchange，指定类型为topicchannel.exchangeDeclare(EXCHANGE_NAME,"topic");//消息内容Stringmessage="新增商品:id=1001";//发送消息，并且指定routingkey为：insert.channel.basicPublish(EXCHANGE_NAME,"aaa.insert",null,message.getBytes());System.out.println("[商品服务：]Sent'"+message+"'");channel.close();connection.close();}}publicclassRecv{privatefinalstaticStringQUEUE_NAME="topic_exchange_queue_2";publicclassRecv{privatefinalstaticStringQUEUE_NAME="topic_exchange_queue_2";privatefinalstaticStringEXCHANGE_NAME="topic_exchange_test";publicstaticvoidmain(String[]argv)throwsException{//获取到连接Connectionconnection=ConnectionUtil.getConnection();//获取通道Channelchannel=connection.createChannel();//声明队列channel.queueDeclare(QUEUE_NAME,false,false,false,null);//绑定队列到交换机，同时指定需要订阅的routingkey。channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"aaa.*");//定义队列的消费者DefaultConsumerconsumer=newDefaultConsumer(channel){//获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用@OverridepublicvoidhandleDelivery(StringconsumerTag,Envelopeenvelope,BasicPropertiesproperties,byte[]body)throwsIOException{//body即消息体Stringmsg=newString(body);System.out.println("[消费者2]received:"+msg+"!");}};//监听队列，自动ACKchannel.basicConsume(QUEUE_NAME,true,consumer);}}2.1.5rabbitMQapi简介连接到一个代理需求参数ConnectionFactoryfactory=newConnectionFactory()ConnectionFactoryfactory=newConnectionFactory();factory.setUsername(userName);factory.setPassword(password);factory.setVirtualHost(virtualHost);factory.setHost(hostName);factory.setPort(portNumber);Connectionconn=factory.newConnection();参数名称说明约束类型备注userName账号必填Stringpassword密码必填StringvirtualHostVhost必填String给此用户分配的权限路径hostname地址必填StringprotNumber端口号必填IntChannelchannel=connection.createChannel();打开通道Channelchannel=connection.createChannel();声明(创建)队列需求参数:channel.queueDeclare(name,durable,autoDelete,exclusive,noWait,args);channel.queueDeclare(name,durable,autoDelete,exclusive,noWait,args);参数名称说明约束类型备注name队列名字必填Stringdurable是否持久化必填BooleanautoDelete是否自动删除队列必填Booleanexclusive是否排外必填Boolean1.是否自动删除2.是否对此队列加锁noWait是否等待服务器返回i必填Booleanargs相关参数必填String一般为null向指定队列中发送消息channel.basicPublish(Echannel.basicPublish(Exchange,routingKey,mandatory,immediate,props,message);参数名称说明约束类型备注Exchange交换器名称必填StringroutingKey路由键必填Stringmandatory非必填Boolean无法找到队列时,是否丢弃immediate非必填Boolean无法找到消费者时,返回给生产者,3.0去掉了这个参数props消息的基本属性必填BasicProperties例如消息头message消息体必填String真正发送的消息监听队列:channel.basicConsume(channel.basicConsume(queue,autoAck,consumerTag,noLocal,exclusive,arguments,callback);需求参数:参数名称说明约束类型备注queue队列名必填StringautoAck是否自动确认消息非必填BooleanconsumerTag消费者标签非必填String区分多个消费者noLocal非必填BooleanTrue:不能将同一个Conenction中生产者发送的消息传递给这个Connection中的消费者exclusive是否排他非必填Booleanarguments消息的基本属性非必填Mapcallback消费者必填ConsumerDefaultConsumer建立使用，重写其中的方法channel.exchangeDeclare(exchangechannel.exchangeDeclare(exchange,type,durable,autodelete,internal,arguments);需求参数:参数名称说明约束类型备注exchange队列名必填Stringtype交换器类型必填BuiltinExchangeType/stringdirect,fanout,topicdurable是否持久化非必填Booleanautodelete是否自动删除非必填Booleaninternal是否内置非必填Booleanarguments其他参数非必填Map比如:alternate-exchange绑定队列到交换机需求参数:channel.queueBind(queuechannel.queueBind(queue,exchange,routingKey,arguments);参数名称说明约束类型备注queue队列名必填Stringexchange交换机必填StringroutingKey路由key必填Stringarguments其他的一些参数非必填Map2.1.6简要规范名称指定系统标识业务系统QueueExchangeCWAA给B发MESSAGEA给B发COMMONWHBB给A发MESSAGEB给A发COMMON规则:管道:QUEUE_发消息的系统_收消息的系统_作用 交换机:EXCHANGE_发消息的系统_收消息的系统_作用3使用手册3.1使用springboot+rabbitMQ收发消息流程3.1.1导入依赖

<<dependency>

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-starter-amqp</artifactId>

<version>1.5.2.RELEASE</version></dependency>3.1.spring.rabbitmq.host=spring.rabbitmq.port=spring.rabbitmq.host=spring.rabbitmq.port=5672spring.rabbitmq.username=guestspring.rabbitmq.password=guest3.1.3第三步：编写RabbitConfig类3.1.4.@Configurationpublic@ConfigurationpublicclassRabbitConfig{

privatefinalLoggerlogger=LoggerFactory.getLogger(this.getClass());

@Value("${spring.rabbitmq.host}")

privateStringhost;

@Value("${spring.rabbitmq.port}")

privateintport;

@Value("${spring.rabbitmq.username}")

privateStringusername;

@Value("${spring.rabbitmq.password}")privateStringpassword;

//交换机名

publicstaticfinalStringEXCHANGE_A="my-mq-exchange_A";//队列机名

publicstaticfinalStringQUEUE_A="QUEUE_A";

//routingKey

publicstaticfinalStringROUTINGKEY_A="spring-boot-routingKey_A";

@Bean

publicConnectionFactoryconnectionFactory(){

CachingConnectionFactoryconnectionFactory=newCachingConnectionFactory(host,port);

connectionFactory.setUsername(username);

connectionFactory.setPassword(password);

connectionFactory.setVirtualHost("/");

connectionFactory.setPublisherConfirms(true);

returnconnectionFactory;

}

@Bean

@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)

//必须是prototype类型

publicRabbitTemplaterabbitTemplate(){

RabbitTemplatetemplate=newRabbitTemplate(connectionFactory());

returntemplate;

}}

@Component@ComponentpublicclassMsgProducerimplementsRabbitTemplate.ConfirmCallback{

privatefinalLoggerlogger=LoggerFactory.getLogger(this.getClass());

//由于rabbitTemplate的scope属性设置为ConfigurableBeanFactory.SCOPE_PROTOTYPE，所以不能自动注入

privateRabbitTemplaterabbitTemplate;

/**

*构造方法注入rabbitTemplate

*/

@Autowired

publicMsgProducer(RabbitTemplaterabbitTemplate){

this.rabbitTemplate=rabbitTemplate;

rabbitTemplate.setConfirmCallback(this);//rabbitTemplate如果为单例的话，那回调就是最后设置的内容

}

publicvoidsendMsg(Stringcontent){

CorrelationDatacorrelationId=newCorrelationData(UUID.randomUUID().toString());

//把消息放入ROUTINGKEY_A对应的队列当中去，对应的是队列A

rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_A,RabbitConfig.ROUTINGKEY_A,content,correlationId);

}

//回调

@Override

publicvoidconfirm(CorrelationDatacorrelationData,booleanack,Stringcause){

("回调id:"+correlationData);

if(ack){

("消息成功消费");

}else{

("消息消费失败:"+cause);}}}3.1./**

*针对消费者配置

*1.设置交换机类型/**

*针对消费者配置

*1.设置交换机类型

*2.将队列绑定到交换机

FanoutExchange:将消息分发到所有的绑定队列，无routingkey的概念

HeadersExchange：通过添加属性key-value匹配

DirectExchange:按照routingkey分发到指定队列

TopicExchange:多关键字匹配

*/

@Bean

publicDirectExchangedefaultExchange(){

returnnewDirectExchange(EXCHANGE_A);

}

/**

*获取队列A

*@return

*/

@Bean

publicQueuequeueA(){

returnnewQueue(QUEUE_A,true);//队列持久

}

@Bean

publicBindingbinding(){

returnBindingBuilder.bind(queueA()).to(defaultExchange()).with(RabbitConfig.ROUTINGKEY_A);

}黄色标注的地方是根据不同的交换机模式绑定不同的交换机3.1.@Component@RabbitListener@Component@RabbitListener(queues=RabbitConfig.QUEUE_A)publicclassMsgReceiver{

privatefinalLoggerlogger=LoggerFactory.getLogger(this.getClass());

@RabbitHandler

publicvoidprocess(Stringcontent){

("接收处理队列A当中的消息："+content);

}}3.2使用spring+rabbitMQ收发消息流程<dependency>

<dependency>

<groupId>com.rabbitmq</groupId>

<artifactId>amqp-client</artifactId>

<version>4.0.2</version></dependency><dependency>

<groupId>org.springframework.amqp</groupId>

<artifactId>spring-rabbit</artifactId>

<version>1.7.9.RELEASE</version></dependency><beans<beansxmlns="/schema/beans"

xmlns:context="/schema/context"

xmlns:xsi="/2001/XMLSchema-instance"xmlns:rabbit="/schema/rabbit"

xsi:schemaLocation="/schema/rabbit

/schema/rabbit/spring-rabbit-1.4.xsd

/schema/beans

/schema/beans/spring-beans-4.1.xsd

/schema/context

/schema/context/spring-context.xsd"></beans>3.配置消息生产者<!--连接工厂--><rabbit:connection-factoryid="rabbitConnectionFactory"host=""port="5672"

<!--连接工厂--><rabbit:connection-factoryid="rabbitConnectionFactory"host=""port="5672"

username="guest"password="guest"

virtual-host="/test"/><!--定义mq管理<!--定义mq管理-->

<rabbit:adminconnection-factory="connectionFactory"/> 3.3定义rabbitmq模板（消息生产者通过模板类进行推送数据）<!--定义rabbitmq模板，指定连接工厂、exchange、queue等--><rabbit:template <!--定义rabbitmq模板，指定连接工厂、exchange、queue等--><rabbit:template id="amqpTemplate" connection-factory="rabbitConnectionFactory"exchange="XXX"queue="XXX"/>消费发送是通过rabbitTemplate.convertAndSend()这个方法进行发送的rabbitTemplate.convertAndSend()用三个参数，分别代表exchange、routing-key（queue）、message，如果调用时不写exchange，第二个参数代表queue，比如rabbitTemplate.convertAndSend(queue,message)表示将消息直接发布到queue队列中如果在模板中指定了默认exchange和queue，如果消息在发布时没指定exchange和queue，则消息直接通过默认的exchange将消息推送给对应的queue，如果只配置了queue，则表示直接将消息发布到queue队列中我通常的做法是不指定exchange和queue，通过代码进行指定

3.4配置队列注意:根据不同的模式配置不同的队列<!--队列声明:<!--队列声明:

durable：true、falsetrue：在服务器重启时，能够存活

exclusive：当连接关闭后是否自动删除队列；是否私有队列，如果私有，其他通道不能访问当前队列

autodelete：当没有任何消费者使用时，自动删除该队列--><!--用于发布/订阅模式的队列--><rabbit:queuename="myFanoutQueue"durable="true"exclusive="false"auto-delete="false"/><!--定义交互机发布/订阅模式--><rabbit:fanout-exchange<!--定义交互机发布/订阅模式--><rabbit:fanout-exchangename="myFanoutExchange"durable="true"auto-delete="false">

<rabbit:bindings>

<rabbit:bindingqueue="myFanoutQueue"></rabbit:binding>

</rabbit:bindings></rabbit:fanout-exchange><!--定义交互机路由模式（需要routing-key）--><rabbit:direct-exchangename="myDirectExchange"durable="true"auto-delete="false">

<rabbit:bindings>

<rabbit:bindingqueue="myDirectQueue"key="error"></rabbit:binding>

</rabbit:bindings></rabbit:direct-exchange>注意:1)标黄的地方,根据不同的模式配置不同的交换机路由模式需要指定key，表示exchange通过key（routing-key）将消息发布到queue中主题模式是通过pattern参数来表示routing-key的,例:"test.#"publicclassSpringSender{

publicstaticvoidsendMessage(Stringexchange,StringroutingKey,Object

message){publicclassSpringSender{

publicstaticvoidsendMessage(Stringexchange,StringroutingKey,Object

message){

//加载配置文件

AbstractApplica