1、項目準備：

使用maven的依賴

<dependencies> <dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>4.0.2</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>1.7.10</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId><version>1.7.5</version></dependency><dependency><groupId>log4j</groupId><artifactId>log4j</artifactId><version>1.2.17</version></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.11</version></dependency></dependencies>

?

?

2、簡單隊列 hello world

2.1、模型圖片

?

P：消息的生產者 C：消息的消費者 紅色：隊列 生產者將消息發送到隊列，消費者從隊列中獲取消息。 那么我們根據以上的模型,咱們抽取出 3 個對象
生產者(用戶發送消息) 隊列(中間件):類似于容器(存儲消息) 消費者(獲取隊列中的消息)

?

2.2、JAVA 操作 獲取 MQ 連接

import java.io.IOException; import java.util.concurrent.TimeoutException; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; public class ConnectionUtils {public static Connection getConnection() throws IOException, TimeoutException{//定義鏈接工廠ConnectionFactory factory = new ConnectionFactory();//設置服務地址factory.setHost("127.0.0.1");//設置端口factory.setPort(5672);//設置賬戶、密碼、vhost等factory.setVirtualHost("/user");factory.setUsername("user");factory.setPassword("user");//通過工程獲取鏈接Connection conn = factory.newConnection();return conn;} }

?

?

2.3、生產者發送數據到消息隊列

import java.io.IOException; import java.util.concurrent.TimeoutException; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.util.ConnectionUtils; public class Send {private static final String QUEUE_NAME ="test_simple_queue";public static void main(String[] args) throws IOException, TimeoutException {//獲取一個連接Connection conn = ConnectionUtils.getConnection();//從消息隊列中獲取一個信道Channel c = conn.createChannel();
//創建隊列聲明c.queueDeclare(QUEUE_NAME, false, false, false, null);//發送信息String msg = "hello simple";c.basicPublish("",QUEUE_NAME , null, msg.getBytes());//關閉資源 c.close();conn.close();} }

?

進行新建隊列的查看：

?

查看相關的消息：

?

2.4、消費者消費

使用較早的api進行對隊列中的消息進行獲取

package com.rabbitmq.simple; import java.io.IOException; import java.util.concurrent.TimeoutException; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConsumerCancelledException; import com.rabbitmq.client.QueueingConsumer; import com.rabbitmq.client.QueueingConsumer.Delivery; import com.rabbitmq.client.ShutdownSignalException; import com.rabbitmq.util.ConnectionUtils; //消費者接受消息 public class Receive {private static final String QUEUE_NAME ="test_simple_queue";public static void main(String[] args) throws IOException, TimeoutException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {//獲取連接Connection conn = ConnectionUtils.getConnection();//創建信道Channel c = conn.createChannel();//定義隊列的消費者QueueingConsumer queueingConsumer = new QueueingConsumer(c);//監聽隊列c.basicConsume(QUEUE_NAME, true,queueingConsumer);while(true){Delivery delivery =queueingConsumer.nextDelivery();String msg = new String(delivery.getBody());System.out.println(msg);}} }

?

?此方法現在已經很少進行使用，如果使用較早的api可以進行了解一下

?

此時的監聽不會關閉！！！ 只要隊列有消息就會進行監聽

?

使用新的api

import java.io.IOException; import java.util.concurrent.TimeoutException; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConsumerCancelledException; import com.rabbitmq.client.DefaultConsumer; import com.rabbitmq.client.Envelope; import com.rabbitmq.client.QueueingConsumer; import com.rabbitmq.client.QueueingConsumer.Delivery; import com.rabbitmq.client.ShutdownSignalException; import com.rabbitmq.client.AMQP.BasicProperties; import com.rabbitmq.util.ConnectionUtils; //消費者接受消息 public class Receive {private static final String QUEUE_NAME ="test_simple_queue";public static void main(String[] args) throws IOException, TimeoutException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {//獲取連接Connection conn = ConnectionUtils.getConnection();//創建頻道Channel channel = conn.createChannel();//隊列聲明channel.queueDeclare(QUEUE_NAME, false, false, false, null);
　　　　　　 //定義一個消費者DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)throws IOException {//重寫父方法String msg = new String(body,"utf-8");System.out.println(msg);}};//進行監聽隊列channel.basicConsume(QUEUE_NAME, true,consumer);} }

?

此時會一致動態的進行對隊列的監控

?

?

3、不足之處

耦合性高 生產消費一一對應(如果有多個消費者想都消費這個消息,就不行了) 隊列名稱變更時需要同時更改

?

轉載于:https://www.cnblogs.com/Mrchengs/p/10530508.html

總結

以上是生活随笔為你收集整理的2、RabbitMQ-simplest thing（简单队列）的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。