目錄

?

?

基本概念

代碼及演示

---

?

基本概念

就是官方的這個(gè)模型：

p為生產(chǎn)者不經(jīng)過交換機(jī)，直接把數(shù)據(jù)傳給消息隊(duì)列，c為consumer用于消費(fèi)。

這種結(jié)構(gòu)在本科生的時(shí)候，經(jīng)常自己寫，現(xiàn)在用RabbitMQ來試試

?

?

代碼及演示

發(fā)送端點(diǎn)擊運(yùn)行：

消費(fèi)者那邊會(huì)接收到數(shù)據(jù)：

關(guān)鍵的源碼如下：

Maven依賴：

<dependencies><dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.4.3</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>1.7.25</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.12</version></dependency></dependencies>

生產(chǎn)者：

package rabbitmq.simple;import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import rabbitmq.util.ConnectionUtils;import java.io.IOException; import java.util.concurrent.TimeoutException;public class Send {private static final String QUEUE_NAME = "test_simple_queue";public static void main(String[] args) throws IOException, TimeoutException {Connection connection = ConnectionUtils.getConnect();Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);String msg = "Hello World";channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());System.out.println("---send msg: " + msg);channel.close();connection.close();} }

?

消費(fèi)者：

package rabbitmq.simple;import com.rabbitmq.client.*; import rabbitmq.util.ConnectionUtils;import java.io.IOException; import java.util.concurrent.TimeoutException;public class Recv {private static final String QUEUE_NAME = "test_simple_queue";public static void main(String[] args) throws IOException, TimeoutException {Connection connection = ConnectionUtils.getConnect();Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null); //聲明，沒有就會(huì)創(chuàng)建DefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String msg = new String(body, "utf-8");System.out.println("recv: " + msg);}};//監(jiān)聽channel.basicConsume(QUEUE_NAME, true, consumer);} }

這里有一點(diǎn)要注意的：

basicConsume里的第二個(gè)參數(shù)是true，的意思是給消息隊(duì)列一個(gè)應(yīng)答，說明，我已經(jīng)拿出來了！他是一個(gè)ack

程序打包下載地址：

https://github.com/fengfanchen/Java/tree/master/SimpleRabbitMQ

總結(jié)

以上是生活随笔為你收集整理的Java笔记-使用RabbitMQ的Java接口生产数据并消费的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

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