RabbitMQ 簡(jiǎn)述#

RabbitMQ是一個(gè)消息代理：它接受并轉(zhuǎn)發(fā)消息。 您可以將其視為郵局：當(dāng)您將要把寄發(fā)的郵件投遞到郵箱中時(shí)，您可以確信Postman 先生最終會(huì)將郵件發(fā)送給收件人。 在這個(gè)比喻中，RabbitMQ是一個(gè)郵箱，郵局和郵遞員，用來(lái)接受，存儲(chǔ)和轉(zhuǎn)發(fā)二進(jìn)制數(shù)據(jù)塊的消息。

隊(duì)列就像是在RabbitMQ中扮演郵箱的角色。 雖然消息經(jīng)過(guò)RabbitMQ和應(yīng)用程序，但它們只能存儲(chǔ)在隊(duì)列中。 隊(duì)列只受主機(jī)的內(nèi)存和磁盤限制的限制，它本質(zhì)上是一個(gè)大的消息緩沖區(qū)。 許多生產(chǎn)者可以發(fā)送到一個(gè)隊(duì)列的消息，許多消費(fèi)者可以嘗試從一個(gè)隊(duì)列接收數(shù)據(jù)。

producer即為生產(chǎn)者，用來(lái)產(chǎn)生消息發(fā)送給隊(duì)列。consumer是消費(fèi)者，需要去讀隊(duì)列內(nèi)的消息。producer，consumer和broker(rabbitMQ server)不必駐留在同一個(gè)主機(jī)上;確實(shí)在大多數(shù)應(yīng)用程序中它們是這樣分布的。

簡(jiǎn)單隊(duì)列#

簡(jiǎn)單隊(duì)列是最簡(jiǎn)單的一種模式，由生產(chǎn)者、隊(duì)列、消費(fèi)者組成。生產(chǎn)者將消息發(fā)送給隊(duì)列，消費(fèi)者從隊(duì)列中讀取消息完成消費(fèi)。

在下圖中，“P”是我們的生產(chǎn)者，“C”是我們的消費(fèi)者。 中間的框是隊(duì)列 - RabbitMQ代表消費(fèi)者的消息緩沖區(qū)。

java 方式#

生產(chǎn)者#

Copypackage com.anqi.mq.nat;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;public class MyProducer { private static final String QUEUE_NAME = "ITEM_QUEUE"; public static void main(String[] args) throws Exception { //1. 創(chuàng)建一個(gè) ConnectionFactory 并進(jìn)行設(shè)置 ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setVirtualHost("/"); factory.setUsername("guest"); factory.setPassword("guest"); //2. 通過(guò)連接工廠來(lái)創(chuàng)建連接 Connection connection = factory.newConnection(); //3. 通過(guò) Connection 來(lái)創(chuàng)建 Channel Channel channel = connection.createChannel(); //實(shí)際場(chǎng)景中，消息多為json格式的對(duì)象 String msg = "hello"; //4. 發(fā)送三條數(shù)據(jù) for (int i = 1; i <= 3 ; i++) { channel.basicPublish("", QUEUE_NAME, null, msg.getBytes()); System.out.println("Send message" + i +" : " + msg); } //5. 關(guān)閉連接 channel.close(); connection.close(); }}Copy /** * Declare a queue * @param queue the name of the queue * @param durable true if we are declaring a durable queue (the queue will survive a server restart) * @param exclusive true if we are declaring an exclusive queue (restricted to this connection) * @param autoDelete true if we are declaring an autodelete queue (server will delete it when no longer in use) * @param arguments other properties (construction arguments) for the queue * @return a declaration-confirm method to indicate the queue was successfully declared * @throws java.io.IOException if an error is encountered */ Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,Map arguments) throws IOException; /** * Publish a message * @see com.rabbitmq.client.AMQP.Basic.Publish * @param exchange the exchange to publish the message to * @param routingKey the routing key * @param props other properties for the message - routing headers etc * @param body the message body * @throws java.io.IOException if an error is encountered */ void basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body) throws IOException; /** * Start a non-nolocal, non-exclusive consumer, with * a server-generated consumerTag. * @param queue the name of the queue * @param autoAck true if the server should consider messages * acknowledged once delivered; false if the server should expect * explicit acknowledgements * @param callback an interface to the consumer object * @return the consumerTag generated by the server * @throws java.io.IOException if an error is encountered * @see com.rabbitmq.client.AMQP.Basic.Consume * @see com.rabbitmq.client.AMQP.Basic.ConsumeOk * @see #basicConsume(String, boolean, String, boolean, boolean, Map, Consumer) */ String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException;

消費(fèi)者#

Copypackage com.anqi.mq.nat;import com.rabbitmq.client.*;import java.io.IOException;public class MyConsumer { private static final String QUEUE_NAME = "ITEM_QUEUE"; public static void main(String[] args) throws Exception { //1. 創(chuàng)建一個(gè) ConnectionFactory 并進(jìn)行設(shè)置 ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setVirtualHost("/"); factory.setUsername("guest"); factory.setPassword("guest"); //2. 通過(guò)連接工廠來(lái)創(chuàng)建連接 Connection connection = factory.newConnection(); //3. 通過(guò) Connection 來(lái)創(chuàng)建 Channel Channel channel = connection.createChannel(); //4. 聲明一個(gè)隊(duì)列 channel.queueDeclare(QUEUE_NAME, true, false, false, null); System.out.println(" [*] Waiting for messages. To exit press CTRL+C"); /* true:表示自動(dòng)確認(rèn)，只要消息從隊(duì)列中獲取，無(wú)論消費(fèi)者獲取到消息后是否成功消費(fèi)，都會(huì)認(rèn)為消息已經(jīng)成功消費(fèi) false:表示手動(dòng)確認(rèn)，消費(fèi)者獲取消息后，服務(wù)器會(huì)將該消息標(biāo)記為不可用狀態(tài)，等待消費(fèi)者的反饋，如果消費(fèi)者一 直沒有反饋，那么該消息將一直處于不可用狀態(tài)，并且服務(wù)器會(huì)認(rèn)為該消費(fèi)者已經(jīng)掛掉，不會(huì)再給其發(fā)送消息， 直到該消費(fèi)者反饋。 */ //5. 創(chuàng)建消費(fèi)者并接收消息 Consumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String message = new String(body, "UTF-8"); System.out.println(" [x] Received '" + message + "'"); } }; //6. 設(shè)置 Channel 消費(fèi)者綁定隊(duì)列 channel.basicConsume(QUEUE_NAME, true, consumer); }}CopySend message1 : helloSend message2 : helloSend message3 : hello [*] Waiting for messages. To exit press CTRL+C [x] Received 'hello' [x] Received 'hello' [x] Received 'hello'

當(dāng)我們啟動(dòng)生產(chǎn)者之后查看RabbitMQ管理后臺(tái)可以看到有一條消息正在等待被消費(fèi)。

當(dāng)我們啟動(dòng)消費(fèi)者之后再次查看，可以看到積壓的一條消息已經(jīng)被消費(fèi)。

總結(jié)#

• 隊(duì)列聲明queueDeclare的參數(shù)：第一個(gè)參數(shù)表示隊(duì)列名稱、第二個(gè)參數(shù)為是否持久化(true表示是，隊(duì)列將在服務(wù)器重啟時(shí)生存)、第三個(gè)參數(shù)為是否是獨(dú)占隊(duì)列(創(chuàng)建者可以使用的私有隊(duì)列，斷開后自動(dòng)刪除)、第四個(gè)參數(shù)為當(dāng)所有消費(fèi)者客戶端連接斷開時(shí)是否自動(dòng)刪除隊(duì)列、第五個(gè)參數(shù)為隊(duì)列的其他參數(shù)。
• basicConsume的第二個(gè)參數(shù)autoAck: 應(yīng)答模式，true：自動(dòng)應(yīng)答，即消費(fèi)者獲取到消息，該消息就會(huì)從隊(duì)列中刪除掉，false：手動(dòng)應(yīng)答，當(dāng)從隊(duì)列中取出消息后，需要程序員手動(dòng)調(diào)用方法應(yīng)答，如果沒有應(yīng)答，該消息還會(huì)再放進(jìn)隊(duì)列中，就會(huì)出現(xiàn)該消息一直沒有被消費(fèi)掉的現(xiàn)象。
• 這種簡(jiǎn)單隊(duì)列的模式，系統(tǒng)會(huì)為每個(gè)隊(duì)列隱式地綁定一個(gè)默認(rèn)交換機(jī)，交換機(jī)名稱為" (AMQP default)"，類型為直連 direct，當(dāng)你手動(dòng)創(chuàng)建一個(gè)隊(duì)列時(shí)，系統(tǒng)會(huì)自動(dòng)將這個(gè)隊(duì)列綁定到一個(gè)名稱為空的 Direct 類型的交換機(jī)上，綁定的路由鍵 routing key 與隊(duì)列名稱相同，相當(dāng)于channel.queueBind(queue:"QUEUE_NAME", exchange:"(AMQP default)“, routingKey:"QUEUE_NAME");雖然實(shí)例沒有顯式聲明交換機(jī)，但是當(dāng)路由鍵和隊(duì)列名稱一樣時(shí)，就會(huì)將消息發(fā)送到這個(gè)默認(rèn)的交換機(jī)中。這種方式比較簡(jiǎn)單，但是無(wú)法滿足復(fù)雜的業(yè)務(wù)需求，所以通常在生產(chǎn)環(huán)境中很少使用這種方式。
• The default exchange is implicitly bound to every queue, with a routing key equal to the queue name. It is not possible to explicitly bind to, or unbind from the default exchange. It also cannot be deleted.默認(rèn)交換機(jī)隱式綁定到每個(gè)隊(duì)列，其中路由鍵等于隊(duì)列名稱。不可能顯式綁定到，或從缺省交換中解除綁定。它也不能被刪除。? ——引自 RabbitMQ 官方文檔?

spring-amqp方式#

引入 Maven 依賴

Copy com.rabbitmq amqp-client 5.6.0org.springframework.amqp spring-rabbit 2.1.5.RELEASE

spring 配置文件

Copy

使用測(cè)試

Copyimport org.springframework.amqp.core.AmqpTemplate;import org.springframework.context.ApplicationContext;import org.springframework.context.support.ClassPathXmlApplicationContext;public class Main { public static void main(String[] args) { ApplicationContext app = new ClassPathXmlApplicationContext("spring/rabbit-context.xml"); AmqpTemplate amqpTemplate = app.getBean(AmqpTemplate.class); amqpTemplate.convertAndSend("MY-QUEUE", "Item"); String msg = (String) amqpTemplate.receiveAndConvert("MY-QUEUE"); System.out.println(msg); }}

參考方法

Copy/** * Convert a Java object to an Amqp {@link Message} and send it to a specific exchange * with a specific routing key. * * @param exchange the name of the exchange * @param routingKey the routing key * @param message a message to send * @throws AmqpException if there is a problem */void convertAndSend(String exchange, String routingKey, Object message) throws AmqpException;/** * Receive a message if there is one from a specific queue and convert it to a Java * object. Returns immediately, possibly with a null value. * * @param queueName the name of the queue to poll * @return a message or null if there is none waiting * @throws AmqpException if there is a problem */@NullableObject receiveAndConvert(String queueName) throws AmqpException;

作者： 海向

出處：https://www.cnblogs.com/haixiang/p/10826710.html

本站使用「CC BY 4.0」創(chuàng)作共享協(xié)議，轉(zhuǎn)載請(qǐng)?jiān)谖恼旅黠@位置注明作者及出處。

總結(jié)

以上是生活随笔為你收集整理的获取rabbitmq连接对象_RabbitMQ——简单队列的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。