以前學習的都是一條消息發給一個消費者,下面開始記錄如何把一條信息發給多個消費者java
這邊咱們用到了交換機Exchangeide
交換機模式:fanout學習
模式特色:生產者把消息發送給Exchange以後,Exchange則會把這些消息添加到與本身綁定的全部隊列之中,監聽這些隊列的消費者就能夠收到這些消息。code
注:Exchange並不能保存信息,若是沒有綁定的隊列,那麼生產者發送數據就會丟失,只有隊列才能存儲生產者的消息。blog
生產者:聲明交換機後指定交換機模式fanoutrabbitmq
package com.example.demo.queue.exchangeToQueue.fanout; import java.io.IOException; import java.util.concurrent.TimeoutException; import com.example.demo.utils.ConnectionUtil; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; public class Producer { // exchange名稱 private static final String EXCHANGE_NAME = "exchange_fanout"; public static void main(String[] args) { Connection connection = null; Channel channel = null; try { // 獲取鏈接 connection = ConnectionUtil.getConnection(); // 建立通道 channel = connection.createChannel(); // 聲明交換機 channel.exchangeDeclare(EXCHANGE_NAME, "fanout", true); // 生產者發送的信息 String msg = "msg from producer:"; for(int i=0;i<10;i++) { msg = "msg from producer :" + i; System.out.println("send msg : "+msg); // 發送信息 channel.basicPublish(EXCHANGE_NAME, "", null, msg.getBytes()); } } catch (Exception e) { e.printStackTrace(); } finally { // 關閉通道 try { channel.close(); } catch (IOException e) { e.printStackTrace(); } catch (TimeoutException e) { e.printStackTrace(); } // 關閉鏈接 try { connection.close(); } catch (IOException e) { e.printStackTrace(); } } } }
消費者1:聲明隊列fanout_exchange_to_queue_01並綁定到交換機exchange_fanout隊列
package com.example.demo.queue.exchangeToQueue.fanout; import java.io.IOException; import com.example.demo.utils.ConnectionUtil; import com.rabbitmq.client.AMQP.BasicProperties; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.DefaultConsumer; import com.rabbitmq.client.Envelope; public class Consumer01 { // 交換機名稱 private static final String EXCHANGE_NAME = "exchange_fanout"; // 監聽隊列名稱 private static final String QUEUE_NAME = "fanout_exchange_to_queue_01"; public static void main(String[] args) { try { // 獲取鏈接 Connection connection = ConnectionUtil.getConnection(); // 建立通道 final Channel channel = connection.createChannel(); // 聲明隊列 channel.queueDeclare(QUEUE_NAME, true, false, false, null); // 綁定隊列到交換機 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ""); // 定義消費者 DefaultConsumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { String msg = new String(body,"UTF-8"); System.out.println("[1]:receive msg:"+msg); System.out.println("[1]:deal msg successful."); } }; // 接收信息 channel.basicConsume(QUEUE_NAME, true, consumer); } catch (Exception e) { e.printStackTrace(); } } }
消費者2:聲明隊列fanout_exchange_to_queue_02並綁定到交換機exchange_fanoutget
package com.example.demo.queue.exchangeToQueue.fanout; import java.io.IOException; import com.example.demo.utils.ConnectionUtil; import com.rabbitmq.client.AMQP.BasicProperties; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.DefaultConsumer; import com.rabbitmq.client.Envelope; public class Consumer02 { // 交換機名稱 private static final String EXCHANGE_NAME = "exchange_fanout"; // 監聽隊列名稱 private static final String QUEUE_NAME = "fanout_exchange_to_queue_02"; public static void main(String[] args) { try { // 獲取鏈接 Connection connection = ConnectionUtil.getConnection(); // 建立通道 final Channel channel = connection.createChannel(); // 聲明隊列 channel.queueDeclare(QUEUE_NAME, true, false, false, null); // 綁定隊列到交換機 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ""); // 定義消費者 DefaultConsumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { String msg = new String(body,"UTF-8"); System.out.println("[2]:receive msg:"+msg); System.out.println("[2]:deal msg successful."); } }; // 接收信息 channel.basicConsume(QUEUE_NAME, true, consumer); } catch (Exception e) { e.printStackTrace(); } } }
消費者3:聲明隊列fanout_exchange_to_queue_03,沒有綁定到交換機exchange_fanoutit
package com.example.demo.queue.exchangeToQueue.fanout; import java.io.IOException; import com.example.demo.utils.ConnectionUtil; import com.rabbitmq.client.AMQP.BasicProperties; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.DefaultConsumer; import com.rabbitmq.client.Envelope; public class Consumer03 { // 交換機名稱 private static final String EXCHANGE_NAME = "exchange_fanout"; // 監聽隊列名稱 private static final String QUEUE_NAME = "fanout_exchange_to_queue_03"; public static void main(String[] args) { try { // 獲取鏈接 Connection connection = ConnectionUtil.getConnection(); // 建立通道 final Channel channel = connection.createChannel(); // 聲明隊列 channel.queueDeclare(QUEUE_NAME, true, false, false, null); // 綁定隊列到交換機 // channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ""); // 定義消費者 DefaultConsumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { String msg = new String(body,"UTF-8"); System.out.println("[3]:receive msg:"+msg); System.out.println("[3]:deal msg successful."); } }; // 接收信息 channel.basicConsume(QUEUE_NAME, true, consumer); } catch (Exception e) { e.printStackTrace(); } } }
此次,涉及到了交換機,而在消費者類中中並無聲明交換機,因此須要先執行生產者類的main方法io
能夠看到生產者已經成功發送消息給交換機,交換機也成功建立了,可是消息數據丟失了,緣由就是上面說的,交換機沒法保存從生產者那接收到的信息,只有隊列能夠保存消息。
接下來,咱們就能夠依次執行三個消費者類的main方法,以後再執行生產者類的main方法。
消費者1終端:
消費者2終端:
消費者3終端: