目錄

?

?

基本概念

代碼與實例

---

?

基本概念

模型如上；

? ? ? ? ? ? 1. 一個生產者，多個消費者；

? ? ? ? ? ? 2. 每個消費者都有自己的隊列；

? ? ? ? ? ? 3. 生產者沒有直接把消息發送到隊列，而是發送到交換機，通過交換機轉發到隊列；

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

? ? ? ? ? ? 5. 生產者發送的消息經過交換機到達隊列就能實現一個消息被多個消費者消費；

?

這里要注意：

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

?

?

代碼與實例

程序運行截圖如下：

生產者：

兩個消費者：

RabbitMQ相關：

關鍵源碼如下：

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);//綁定到交換機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();//聲明交換機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

總結

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

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