在RabbitMQ中，producer只能將消息發送到一個exchange中。要理解exchange也非常簡單，它一邊負責接收producer發送的消息， 另一邊將消息推送到queue中。exchange必須清楚的知道在收到消息之后該如何進行下一步的處理，比如是否應該將這條消息發送到某個queue中？ 還是應該發送到多個queue中？還是應該直接丟棄這條消息等等。

RabbitMQ中的exchange類型有這么幾種：direct，topic，headers以及fanout。這一小節將會主要介紹最后一種類型——fanout。 使用RabbitMQ的client來創建一個fanout類型的exchange。fanout類型的exchange非常簡單，從名字也可以猜測出來，它會向所有的queue廣播所有收到的消息。這正是我們的log系統需要的。

package com.rabbitmq.www.publish_subscribe;import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.Connection; import com.rabbitmq.client.Channel;public class EmitLog {private static final String EXCHANGE_NAME = "logs";private final static String HOST_ADDR = "172.18.112.102";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost(HOST_ADDR);Connection connection = factory.newConnection();Channel channel = connection.createChannel();//定義exchange channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);for(int i=0;i<=10;i++){String message = "helloworld"+i;//信息發送給申明的exchangechannel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));System.out.println(" [x] Sent '" + message );}channel.close();connection.close();}} package com.rabbitmq.www.publish_subscribe;import com.rabbitmq.client.*;import java.io.IOException;public class ReceiveLogs {private static final String EXCHANGE_NAME = "logs";private final static String HOST_ADDR = "172.18.112.102";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost(HOST_ADDR);Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);//服務器為創建隨機queue提供了一個無參數的queueDeclare()方來來創建一個非持久化的、獨有的并且是自動刪除的已命名的queue。String queueName = channel.queueDeclare().getQueue();//現在我們需要告訴exchange將消息發送到我們的queue中。 這種exchange和queue的關系稱為綁定(binding)。channel.queueBind(queueName, EXCHANGE_NAME, "");System.out.println(" [*] Waiting for messages. To exit press CTRL+C");Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties, byte[] body) throws IOException {String message = new String(body, "UTF-8");System.out.println(" [x] Received '" + message + "'");}};channel.basicConsume(queueName, true, consumer);} }

?

轉載于:https://www.cnblogs.com/woms/p/7040857.html

總結

以上是生活随笔為你收集整理的Publish/Subscribe的全部內容，希望文章能夠幫你解決所遇到的問題。

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