1、項目準(zhǔn)備：

使用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：消息的生產(chǎn)者 C：消息的消費者 紅色：隊列 生產(chǎn)者將消息發(fā)送到隊列，消費者從隊列中獲取消息。 那么我們根據(jù)以上的模型,咱們抽取出 3 個對象
生產(chǎn)者(用戶發(fā)送消息) 隊列(中間件):類似于容器(存儲消息) 消費者(獲取隊列中的消息)

?

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();//設(shè)置服務(wù)地址factory.setHost("127.0.0.1");//設(shè)置端口factory.setPort(5672);//設(shè)置賬戶、密碼、vhost等factory.setVirtualHost("/user");factory.setUsername("user");factory.setPassword("user");//通過工程獲取鏈接Connection conn = factory.newConnection();return conn;} }

?

?

2.3、生產(chǎn)者發(fā)送數(shù)據(jù)到消息隊列

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();
//創(chuàng)建隊列聲明c.queueDeclare(QUEUE_NAME, false, false, false, null);//發(fā)送信息String msg = "hello simple";c.basicPublish("",QUEUE_NAME , null, msg.getBytes());//關(guān)閉資源 c.close();conn.close();} }

?

進(jìn)行新建隊列的查看：

?

查看相關(guān)的消息：

?

2.4、消費者消費

使用較早的api進(jìn)行對隊列中的消息進(jìn)行獲取

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();//創(chuàng)建信道Channel c = conn.createChannel();//定義隊列的消費者QueueingConsumer queueingConsumer = new QueueingConsumer(c);//監(jiān)聽隊列c.basicConsume(QUEUE_NAME, true,queueingConsumer);while(true){Delivery delivery =queueingConsumer.nextDelivery();String msg = new String(delivery.getBody());System.out.println(msg);}} }

?

?此方法現(xiàn)在已經(jīng)很少進(jìn)行使用，如果使用較早的api可以進(jìn)行了解一下

?

此時的監(jiān)聽不會關(guān)閉！！！ 只要隊列有消息就會進(jìn)行監(jiān)聽

?

使用新的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();//創(chuàng)建頻道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);}};//進(jìn)行監(jiān)聽隊列channel.basicConsume(QUEUE_NAME, true,consumer);} }

?

此時會一致動態(tài)的進(jìn)行對隊列的監(jiān)控

?

?

3、不足之處

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

?

轉(zhuǎn)載于:https://www.cnblogs.com/Mrchengs/p/10530508.html

總結(jié)

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

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