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

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

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;//信息發(fā)送給申明的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);//服務(wù)器為創(chuàng)建隨機(jī)queue提供了一個(gè)無參數(shù)的queueDeclare()方來來創(chuàng)建一個(gè)非持久化的、獨(dú)有的并且是自動(dòng)刪除的已命名的queue。String queueName = channel.queueDeclare().getQueue();//現(xiàn)在我們需要告訴exchange將消息發(fā)送到我們的queue中。 這種exchange和queue的關(guān)系稱為綁定(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);} }

?

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

總結(jié)

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

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