归档5.消息中间件rabbitmq讲义研究

RabbitMQQ全称为sseQue，即消息队列，RitQ是由rlang语言开发，基于AP（Avacdessaeuee高级消息队列协议）协议实现的消息队列，它是一种应用程序之间的通信方法，消息队列在分布式系统开发中应用非常广泛。RabitMQ地址：htp: / MQRedis5、SpringBoot默认已集成RabAMQP总结：AMP是一套公开的消息队列协议，最早在200年被提出，它旨在从协议层定义消息通信数据的标准格式，为的就是解决M市场上协议不统一的问题。RabitMQ就是遵循AP标准协议开发的M服务。：JS是jav提供的一套消息服务AP标准，其目的是为所有的jva应用程序提供统的消息通信的标准，类似jv的jd，只要遵循j标准的应用程序之间都可以进行消息通信。它和AQ有什么不同，js是jav语言专属的消息服务标准，它是在api层定义标准，并且只能用于jv应用；而AMP是在协议层定义的标准，是跨语言的。22Producer：消息生产者，即生产方客户端，生产方客户端将消息发送到MQ。4、Exchange将消息转发到指定的Queue（队列2.2.1安RabbitMQ由Erlang语言开发Erlang语言用于并发及分布式系统的开发，在电信领域应用广泛，OTP（Open要安装Erlang/OTP 本项目使用Erlang/OTP20.3版本和RabbitMQ3.7.3win6420.3.exe或去老师提供的软件包中找到 erlang安装完成需要配置erlang环境变量：ERLANG_HOME=D:\ProgramFiles\erl9.3在path中添

/rabbitmq/rabbitmq-或去老师提供的软件包中找 2.2.2RabbitMQService-install:RabbitMQService-remove删除服务RabbitMQService-start启动RabbitMQService-stop启动如果没有开始菜单则进入安 下 rabbitmq-service.batinstall安装服务rabbitmq-service.batstop停止服务rabbitmq-service.batstart管理员运行rabbitmq-plugins.batenable3、启动成功登录2.2.3o按照 java我们先用rabbitMQ提供的java 参考： /rabitmq/rabbitmq-java- 创建生产者工程和消费者工程，分别加入RabbitMQ 的依赖<version>4.0.3</version><!‐‐<version>4.0.3</version><!‐‐此版本与springboot1.5.9版本匹配factory.setPassword("g 队 param1：Exchange的名称，如果没有指定，则使用Defaultcatch(Exception @paramconsumerTag消费者 @paramenvelope消息包的内容，可从中获取消息id，消息routingkeypublicvoidhandleDelivery(StringEnvelopeenvelope,AMQP.BasicPropertiesproperties,byte[]body)throwsIOExceptionStringexchange=StringroutingKey=longdeliveryTag=Stringmsg=newSystem.out.println("receivemessage.."+}队列Stringqueue,booleanautoAck,Consumer channel.basicConsume(QUEUE,true,}}２.2.4RabbitMQ有以下几种工作模式：1、Workworkqueues与程序相比，多了一个消费端，两个消费端共同消费同一个队列中的消息。应用场景：对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。、生产者将消息发给be，由交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收到消息用户通知，当用户充值成功或转账完成系统通知用户，通知方式有、邮件多种方法Exchange_fanout_inform两个队列并且绑定到此交换机，绑定时不需要指定otingy发送消息时不需要指定outingypackagepackageprivatestaticfinalStringQUEUE_INFORM_ ="queue_inform_ privatestaticfinalStringQUEUE_INFORM_SMS="queue_inform_sms";privatestaticfinalStringEXCHANGE_FANOUT_INFORM="exchange_fanout_inform";publicstaticvoidmain(String[]args){Connectionconnection=null;Channelchannel=null;tryConnectionFactoryfactory=newConnectionFactory();器 connection=channel=connection.createChannel(); 交换机Stringexchange,BuiltinExchangType channel.exchangeDeclare(EXCHANGE_FANOUT_INORM,B (Stringqueue,booleandurable,booleanexclusive,booleanautoDelete,Map<String,Object>arguments)4 ,true,false,false,null);channel.queueDclare(QUEUE_INFORM_SMS,true,false,false,null);//交换机和队列绑定Stringqueue,Stringexchange,String for(intStringmessage="informto//向交换机发送消息Stringexchange,StringroutingKey,BasicProperties ：byte[]byte[]1、交换机名称，不指令使用默认交换机名称Defaultchannel.basicPublish(EXCHANGE_FANOUT_INFORM,"",null,message.getBytes());System.out.println("SendMessageis:'"+message+"'");}}catch(IOException{;}catch(TimeoutException{try{}catch(IOException{}catch(TimeoutException{}}try{}catch(IOExceptio{}}}}packagepackage@author@version@create2018‐06‐14publicclass privatestaticfinalStringQUEUE_INFORM_ ="inform_queue_ privatestaticfinalStringEXCHANGE_FANOUT_INFORM="inform_exchange_fanout";publicstaticvoidmain(String[]args)throwsIOException,TimeoutExceptionConnectionFactoryfactory=newConnectionFactory(); Connectionconnection=//Channelchannel= 交换机Stringexchange,BuiltinExhangeTyp2、交换机类型，fanout、topicdirectchannel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, channel.queueDeclare(Strngqueue,booleandurable,booleanexclusive,booleanautoDelete,Map<String,Object>arguments) ,true,false,false,//交换机和队列绑定Stringqueue,Stringexchange,String DefaultConsumerdefaultConsumer=new{ ,true,defaultCons 1、publish/subscribe与workqueues有什么区别。workqueues不用定义交换机，而publishsubscribepulish/subscrib的生产方是面向交换机发送消息，workqees的生产方是面向队列发送消息(pulish/subscrib需要设置队列和交换机的绑定，wrkuus不需要设置，实质上wokqees。2、实质工作用什么publish/subscribe还是workqueues建议使用pulish/subscribe，发布订阅模式比工作队列模式更强大，并且发布订阅模式可以指定自己的交换机。exchange_routing_inform两个队列并且绑定到此交换机，绑定时需要指定otingy发送消息时需要指定outingyimportimport器 connection=channel=connection.createChannel(); 交换机Stringexchange,BuiltinExchangeTypechannel.exchangeDeclare(EXCHANGE_ROUTING_INFORM,BuiltinExch channel.queueDeclare(Stringqueue,booleandurable,boleaexcluive,booleanautoDelete,Map<String,Object>arguments) ,true,false,false,null);channel.queueDeclare(QUEUEINFORM_SMS,true,false,false,null);//交换机和队列绑定Stringqueue,Stringxchange,String3路由channel.qeueBind(QUEUE_INFORM_ for(intStringmessage= informto//向交换机发送消息Stringexchange,StringroutingKey,BasicPropertiesbyte[]

1、交换机名称，不指令使用默认交换机名称Default , System.out.println("SendMessageis:'"+message+byte[]

for(intStringmessage="smsinformto//向交换机发送消息Stringexchange,StringroutingKey,BasicProperties：System.out.println("SendMessageis:'"+message+}}catch(IOException{;}catch(TimeoutException{try{}catch(IOException{}catch(TimeoutException{}}try{}catch(IOExceptio{}}}}}packagepackageimportpublicclass privatestaticfinalStringQUEUE_INFORM_ ="inform_queue_ privatestaticfinalStringEXCHANGE_ROUTING_INFORM="inform_exchange_routing";publicstaticvoidmain(String[]args)throwsIOException,TimeoutExceptionConnectionFactoryfactory=newConnectionFactory(); Connectionconnection=//Channelchannel= 交换机Stringexchange,BuiltinExchangeType2、交换机类型，fanout、topic、direct、headchannel.exchangeDeclare(EXCHAGE_ROUTING_INFORM, channel.queueDeclare(Stringqueuebooleandurable,booleanexclusive,booleanautoDelete,Map<String,Object>argments)channel.queueDecla ,true,false,false,//交换机和队列绑定Stringqueue,Stringexchange,String DefaultConsumerdefaultConsumer=new{publicvoidhandleDelivery(StringconsumerTag,Envelope ,true,4.4.41、Routing模式和Publish/subscibe根据用户设置去通知用户，设置接收 param2:交换机类型四种交换机类型：direct、fanout、topic、channel.exchangeDeclare(EXCHANGE_TOPICS_INFORM, ",null,"inform.sms",null,",null,importimport接channel.exchangeDeclare(EXCHANGE_TOPICS_INFORM, ,true,false,false,null);channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null);byte[]

for(intStringmessage= informto//向交换机发送消息Stringexchange,StringroutingKey,BaicProperties1、交换机名称，不指令使用默认交换机名称DefaultE

",：：System.out.println("SedMessageis:'"+message+} fo(inti=0;i<10;Stringmessage="smsinformtocnnel.basicPublh(EXCHANGE_TOPICS_INFORM,"inform.sms",null,System.out.println("SendMessageis:'"+message+} for(intStringmessage="smsand informtouser"+i;channel.basicPublish(EXCHANGE_TOPICS_INFORM,"inform.sms. ",null,System.out.println("SendMessageis:'"+message+}}catch(IOException{}catch(TimeoutException{ channel.exchangeDeclare(EXCHANGE_TOPICS_INFORM,使用Routing模式也可以实现本案例，共设置三个rouingkey，分别 、sms、 队列绑 Headerheadr模式与un不同的地方在于，hedr模式取消outingy，使用headr中的y/vle（键值对）队列。根据用户设置去通知用户，设置接收 channel.basicPublish(EXCHANGE_HEADERS_INFORM,"",properties.bild(,,EXCHANGE_HRPC即客户端调用服务端的方法，使用MQ可以实现RPC的异步调用，基于Direct交换机实现，流程如下Spring整合搭建SpringBoot我们选择基于Spring-Rabbit去操作/spring-projects/spring-配置连接rabbitmq、定义RabitCofi类，配置Echange、Queeoi交换机。importimportpublicclassRabbitmqConfigpublics