目錄

?

?

基本概念

代碼與實(shí)例

---

?

基本概念

模型如上；

? ? ? ? ? ? 1. 一個生產(chǎn)者，多個消費(fèi)者；

? ? ? ? ? ? 2. 每個消費(fèi)者都有自己的隊列；

? ? ? ? ? ? 3. 生產(chǎn)者沒有直接把消息發(fā)送到隊列，而是發(fā)送到交換機(jī)，通過交換機(jī)轉(zhuǎn)發(fā)到隊列；

? ? ? ? ? ? 4. 每個隊列都要綁定到交換機(jī)上；

? ? ? ? ? ? 5. 生產(chǎn)者發(fā)送的消息經(jīng)過交換機(jī)到達(dá)隊列就能實(shí)現(xiàn)一個消息被多個消費(fèi)者消費(fèi)；

?

這里要注意：

? ? ? ? ? ? 1. 在RabbitMQ中交換機(jī)沒有存儲能力，只有隊列里面有；

?

?

代碼與實(shí)例

程序運(yùn)行截圖如下：

生產(chǎn)者：

兩個消費(fèi)者：

RabbitMQ相關(guān)：

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

RecvX.java

package ps;import com.rabbitmq.client.*; import util.ConnectionUtils;import java.io.IOException; import java.util.concurrent.TimeoutException;public class Recv1 {private static final String QUEUE_NAME = "test_queue_fanout_sms";private static final String EXCHAGE_NAME = "test_exchange_fanout";public static void main(String[] args) throws IOException, TimeoutException {Connection connection = ConnectionUtils.getConnect();final Channel channel = connection.createChannel();//聲明隊列channel.queueDeclare(QUEUE_NAME, false, false, false, null);//綁定到交換機(jī)channel.queueBind(QUEUE_NAME, EXCHAGE_NAME, "");channel.basicQos(1);Consumer 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("[1] Recv msg : " + msg);try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();} finally {System.out.println("[1] done");channel.basicAck(envelope.getDeliveryTag(), false);}}};boolean autoAck = false;channel.basicConsume(QUEUE_NAME, autoAck, consumer);} }

Send.java

package ps;import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import util.ConnectionUtils;import java.io.IOException; import java.util.concurrent.TimeoutException;public class Send {private static final String EXCHANGE_NAME = "test_exchange_fanout";public static void main(String[] args) throws IOException, TimeoutException {Connection connection = ConnectionUtils.getConnect();Channel channel = connection.createChannel();//聲明交換機(jī)channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); //并分配String msg = "Hello ps";channel.basicPublish(EXCHANGE_NAME, "", null, msg.getBytes());System.out.println("Send : " + msg);channel.close();connection.close();} }

ConnectionUtils.java

package util;import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory;import java.io.IOException; import java.util.concurrent.TimeoutException;public class ConnectionUtils {public static Connection getConnect() throws IOException, TimeoutException {ConnectionFactory factory = new ConnectionFactory();factory.setHost("127.0.0.1");factory.setPort(5672);factory.setVirtualHost("/vhost_cff");factory.setUsername("cff");factory.setPassword("123");return factory.newConnection();} }

源碼下載地址：

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

總結(jié)

以上是生活随笔為你收集整理的Java笔记-使用RabbitMQ的Java接口实现Publish/Subscribe(订阅模式)的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

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