RabbitMQ简单队列实例及原理解析

2023-08-04 18:51:03 90

这篇文章主要介绍了RabbitMQ简单队列实例及原理解析,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下

RabbitMQ简述

RabbitMQ是一个消息代理：它接受并转发消息。您可以将其视为邮局：当您将要把寄发的邮件投递到邮箱中时，您可以确信Postman先生最终会将邮件发送给收件人。在这个比喻中，RabbitMQ是一个邮箱，邮局和邮递员，用来接受，存储和转发二进制数据块的消息。

队列就像是在RabbitMQ中扮演邮箱的角色。虽然消息经过RabbitMQ和应用程序，但它们只能存储在队列中。队列只受主机的内存和磁盘限制的限制，它本质上是一个大的消息缓冲区。许多生产者可以发送到一个队列的消息，许多消费者可以尝试从一个队列接收数据。

producer即为生产者，用来产生消息发送给队列。consumer是消费者，需要去读队列内的消息。producer，consumer和broker(rabbitMQserver)不必驻留在同一个主机上;确实在大多数应用程序中它们是这样分布的。

简单队列

简单队列是最简单的一种模式，由生产者、队列、消费者组成。生产者将消息发送给队列，消费者从队列中读取消息完成消费。

在下图中，“P”是我们的生产者，“C”是我们的消费者。中间的框是队列-RabbitMQ代表消费者的消息缓冲区。

java方式

生产者

packagecom.anqi.mq.nat;

importcom.rabbitmq.client.Channel;
importcom.rabbitmq.client.Connection;
importcom.rabbitmq.client.ConnectionFactory;

publicclassMyProducer{
privatestaticfinalStringQUEUE_NAME="ITEM_QUEUE";

publicstaticvoidmain(String[]args)throwsException{
//1.创建一个ConnectionFactory并进行设置
ConnectionFactoryfactory=newConnectionFactory();
factory.setHost("localhost");
factory.setVirtualHost("/");
factory.setUsername("guest");
factory.setPassword("guest");

//2.通过连接工厂来创建连接
Connectionconnection=factory.newConnection();

//3.通过Connection来创建Channel
Channelchannel=connection.createChannel();

//实际场景中，消息多为json格式的对象
Stringmsg="hello";
//4.发送三条数据
for(inti=1;i<=3;i++){
channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());
System.out.println("Sendmessage"+i+":"+msg);
}

//5.关闭连接
channel.close();
connection.close();
}
}

/**
*Declareaqueue
*@paramqueuethenameofthequeue
*@paramdurabletrueifwearedeclaringadurablequeue(thequeuewillsurviveaserverrestart)
*@paramexclusivetrueifwearedeclaringanexclusivequeue(restrictedtothisconnection)
*@paramautoDeletetrueifwearedeclaringanautodeletequeue(serverwilldeleteitwhennolongerinuse)
*@paramargumentsotherproperties(constructionarguments)forthequeue
*@returnadeclaration-confirmmethodtoindicatethequeuewassuccessfullydeclared
*@throwsjava.io.IOExceptionifanerrorisencountered
*/
Queue.DeclareOkqueueDeclare(Stringqueue,booleandurable,booleanexclusive,booleanautoDelete,Maparguments)throwsIOException;

/**
*Publishamessage
*@seecom.rabbitmq.client.AMQP.Basic.Publish
*@paramexchangetheexchangetopublishthemessageto
*@paramroutingKeytheroutingkey
*@parampropsotherpropertiesforthemessage-routingheadersetc
*@parambodythemessagebody
*@throwsjava.io.IOExceptionifanerrorisencountered
*/
voidbasicPublish(Stringexchange,StringroutingKey,BasicPropertiesprops,byte[]body)throwsIOException;


/**
*Startanon-nolocal,non-exclusiveconsumer,with
*aserver-generatedconsumerTag.
*@paramqueuethenameofthequeue
*@paramautoAcktrueiftheservershouldconsidermessages
*acknowledgedoncedelivered;falseiftheservershouldexpect
*explicitacknowledgements
*@paramcallbackaninterfacetotheconsumerobject
*@returntheconsumerTaggeneratedbytheserver
*@throwsjava.io.IOExceptionifanerrorisencountered
*@seecom.rabbitmq.client.AMQP.Basic.Consume
*@seecom.rabbitmq.client.AMQP.Basic.ConsumeOk
*@see#basicConsume(String,boolean,String,boolean,boolean,Map,Consumer)
*/
StringbasicConsume(Stringqueue,booleanautoAck,Consumercallback)throwsIOException;

消费者

packagecom.anqi.mq.nat;

importcom.rabbitmq.client.*;
importjava.io.IOException;

publicclassMyConsumer{

privatestaticfinalStringQUEUE_NAME="ITEM_QUEUE";

publicstaticvoidmain(String[]args)throwsException{
//1.创建一个ConnectionFactory并进行设置
ConnectionFactoryfactory=newConnectionFactory();
factory.setHost("localhost");
factory.setVirtualHost("/");
factory.setUsername("guest");
factory.setPassword("guest");

//2.通过连接工厂来创建连接
Connectionconnection=factory.newConnection();

//3.通过Connection来创建Channel
Channelchannel=connection.createChannel();

//4.声明一个队列
channel.queueDeclare(QUEUE_NAME,true,false,false,null);
System.out.println("[*]Waitingformessages.ToexitpressCTRL+C");

/*
true:表示自动确认，只要消息从队列中获取，无论消费者获取到消息后是否成功消费，都会认为消息已经成功消费
false:表示手动确认，消费者获取消息后，服务器会将该消息标记为不可用状态，等待消费者的反馈，如果消费者一
直没有反馈，那么该消息将一直处于不可用状态，并且服务器会认为该消费者已经挂掉，不会再给其发送消息，
直到该消费者反馈。
*/

//5.创建消费者并接收消息
Consumerconsumer=newDefaultConsumer(channel){
@Override
publicvoidhandleDelivery(StringconsumerTag,Envelopeenvelope,
AMQP.BasicPropertiesproperties,byte[]body)
throwsIOException{
Stringmessage=newString(body,"UTF-8");
System.out.println("[x]Received'"+message+"'");
}
};

//6.设置Channel消费者绑定队列
channel.basicConsume(QUEUE_NAME,true,consumer);

}
}

Sendmessage1:hello
Sendmessage2:hello
Sendmessage3:hello

[*]Waitingformessages.ToexitpressCTRL+C
[x]Received'hello'
[x]Received'hello'
[x]Received'hello'

当我们启动生产者之后查看RabbitMQ管理后台可以看到有一条消息正在等待被消费。

当我们启动消费者之后再次查看，可以看到积压的一条消息已经被消费。

总结

队列声明queueDeclare的参数：第一个参数表示队列名称、第二个参数为是否持久化（true表示是，队列将在服务器重启时生存）、第三个参数为是否是独占队列（创建者可以使用的私有队列，断开后自动删除）、第四个参数为当所有消费者客户端连接断开时是否自动删除队列、第五个参数为队列的其他参数。
basicConsume的第二个参数autoAck:应答模式，true：自动应答，即消费者获取到消息，该消息就会从队列中删除掉，false：手动应答，当从队列中取出消息后，需要程序员手动调用方法应答，如果没有应答，该消息还会再放进队列中，就会出现该消息一直没有被消费掉的现象。
这种简单队列的模式，系统会为每个队列隐式地绑定一个默认交换机，交换机名称为"(AMQPdefault)"，类型为直连direct，当你手动创建一个队列时，系统会自动将这个队列绑定到一个名称为空的Direct类型的交换机上，绑定的路由键routingkey与队列名称相同，相当于channel.queueBind(queue:"QUEUE_NAME",exchange:"(AMQPdefault)“,routingKey:"QUEUE_NAME");虽然实例没有显式声明交换机，但是当路由键和队列名称一样时，就会将消息发送到这个默认的交换机中。这种方式比较简单，但是无法满足复杂的业务需求，所以通常在生产环境中很少使用这种方式。
Thedefaultexchangeisimplicitlyboundtoeveryqueue,witharoutingkeyequaltothequeuename.Itisnotpossibletoexplicitlybindto,orunbindfromthedefaultexchange.Italsocannotbedeleted.默认交换机隐式绑定到每个队列，其中路由键等于队列名称。不可能显式绑定到，或从缺省交换中解除绑定。它也不能被删除。 ——引自RabbitMQ官方文档

spring-amqp方式

引入Maven依赖


com.rabbitmq
amqp-client
5.6.0


org.springframework.amqp
spring-rabbit
2.1.5.RELEASE

spring配置文件

使用测试

importorg.springframework.amqp.core.AmqpTemplate;
importorg.springframework.context.ApplicationContext;
importorg.springframework.context.support.ClassPathXmlApplicationContext;

publicclassMain{
publicstaticvoidmain(String[]args){
ApplicationContextapp=newClassPathXmlApplicationContext("spring/rabbit-context.xml");
AmqpTemplateamqpTemplate=app.getBean(AmqpTemplate.class);
amqpTemplate.convertAndSend("MY-QUEUE","Item");
Stringmsg=(String)amqpTemplate.receiveAndConvert("MY-QUEUE");
System.out.println(msg);
}
}

参考方法

/**
*ConvertaJavaobjecttoanAmqp{@linkMessage}andsendittoaspecificexchange
*withaspecificroutingkey.
*
*@paramexchangethenameoftheexchange
*@paramroutingKeytheroutingkey
*@parammessageamessagetosend
*@throwsAmqpExceptionifthereisaproblem
*/
voidconvertAndSend(Stringexchange,StringroutingKey,Objectmessage)throwsAmqpException;
/**
*ReceiveamessageifthereisonefromaspecificqueueandconvertittoaJava
*object.Returnsimmediately,possiblywithanullvalue.
*
*@paramqueueNamethenameofthequeuetopoll
*@returnamessageornullifthereisnonewaiting
*@throwsAmqpExceptionifthereisaproblem
*/
@Nullable
ObjectreceiveAndConvert(StringqueueName)throwsAmqpException;

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持毛票票。

声明：本文内容来源于网络，版权归原作者所有，内容由互联网用户自发贡献自行上传，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任。如果您发现有涉嫌版权的内容，欢迎发送邮件至：czq8825#qq.com（发邮件时，请将#更换为@）进行举报，并提供相关证据，一经查实，本站将立刻删除涉嫌侵权内容。

RabbitMQ简单队列实例及原理解析

热门推荐

随机推荐