? ? ?翻譯地址：http://www.rabbitmq.com/tutorials/tutorial-five-java.html ? ? ?在上一篇博文中，我們進一步改良了日志系統。使用Direct類型的轉換器，使得接收者有能力進行選擇性的接收日志,，而非fanout那樣，只能夠無腦的轉發，如果你還不了解，請閱讀：[輕松搞定RabbitMQ（四）——發布/訂閱](http://blog.csdn.net/xiaoxian8023/article/details/48729479)。 ? ? ?雖然使用Direct類型的轉換器改進了日志系統。但它仍然有一定的局限性——不能根據多重條件進行路由選擇。 ? ? ?在我們的日志系統中，我們可能不僅僅根據日志嚴重性訂閱日志，也想根據發送源訂閱。你可能從unix工具syslog了解過這個概念，它可以根據嚴重性（info/warning/crit…）和設備（auth/cron/kern…）轉發日志。 ? ? ?這將給我們更多的靈活性——我們可以訂閱來自元‘cron’的致命錯誤日志，同時也可以訂閱‘kern’的所有日志。 ? ? ?為了在我們的日志系統中實現上述需求，我們需要了解更復雜的主題類型的轉發器——Topic Exchange。 # Topic exchange（主題轉發器） ? ? ?發送給主題轉發器的消息不能是任意設置的選擇鍵，必須是用小數點隔開的一系列的標識符。這些標識符可以是隨意，但是通常跟消息的某些特性相關聯。一些合法的路由選擇鍵比如“socket.usd.nyse”,"nyse.vmw","quick.orange.rabbit"，你愿意用多少單詞都可以，只要不超過上限的255個字節。 ? ? ?綁定鍵也必須以相同的格式。主題轉發器的邏輯類似于direct類型的轉發器。消息通過一個特定的路由鍵發送到所有與綁定鍵匹配的隊列中。需要注意的是，關于綁定鍵有兩種特殊的情況：*（星號）可以代替任意一個標識符 ；#（井號）可以代替零個或多個標識符。 ? ? ?舉一個簡單例子，如下圖：  ? ? ?在上圖例子中，我們發送描述動物的消息。消息會轉發給包含3個單詞（2個小數點）的路由鍵綁定的隊列中。綁定鍵中的第一個單詞描述的是速度，第二個是顏色，第三個是物種：“速度.顏色.物種”。 ? ? ?我們創建3個綁定：Q1綁定鍵是“*.orange.*”，Q2綁定鍵是“*.*.rabbit”，Q3綁定鍵是“lazy.#”。這些綁定可以概括為：Q1只對橙色的動物感興趣。Q2則是關注兔子和所有懶的動物。 ? ? ?路由鍵為“quick.orange.rabbit”的消息會被路由到2個隊列中去。而“lazy.orange.elephant”的消息同樣會發往2個隊列。另外“quick.orange.fox” 僅僅發往第一個隊列，而"lazy.brown.fox"則只發往第二個隊列。“quick.brown.fox”則所有的綁定鍵都不匹配而被丟棄。 ? ? ?如果我們違反約定，發送了只帶1個或者4個標識符的選擇鍵，像“orange”或者“quick.orange.male.rabbit”，會發生什么呢？這些消息都不匹配任何綁定，所以將被丟棄。 ? ? ?另外，“lazy.orange.male.rabbit”，盡管有4個標識符，但是仍然匹配最后一個綁定鍵，所以會發送到第二個隊列中。 ? ? ?注：主題類型的轉發器非常強大，可以實現其他類型的轉發器。當隊列綁定#綁定鍵，可以接受任何消息，類似于fanout轉發器。當特殊字符*和#不包含在綁定鍵中，這個主題轉發器就像一個direct類型的轉發器。 # 完整實例 ? ? ?我們將主題類型的轉發器應用到日志系統中，路由格式為：“<設備>.<嚴重級別>”。 發送端（EmitLogTopic.java） ~~~ public class EmitLogDirect { private final static String EXCHANGE_NAME = "topic_logs"; public static void main(String[] args) throws IOException { /** * 創建連接連接到MabbitMQ */ ConnectionFactory factory = new ConnectionFactory(); // 設置MabbitMQ所在主機ip或者主機名 factory.setHost("127.0.0.1"); // 創建一個連接 Connection connection = factory.newConnection(); // 創建一個頻道 Channel channel = connection.createChannel(); // 指定轉發——廣播 channel.exchangeDeclare(EXCHANGE_NAME, "topic"); //所有設備和日志級別 String[] facilities ={"auth","cron","kern","auth.A"}; String[] severities={"error","info","warning"}; for(int i=0;i<4;i++){ for(int j=0;j<3;j++){ //每一個設備，每種日志級別發送一條日志消息 String routingKey = facilities[i]+"."+severities[j%3]; // 發送的消息 String message =" Hello World!"+Strings.repeat(".", i+1); //參數1：exchange name //參數2：routing key channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes()); System.out.println(" [x] Sent [" + routingKey +"] : '"+ message + "'"); } } // 關閉頻道和連接 channel.close(); connection.close(); } } ~~~ 消費者1（ReceiveLogs2Console.java） ~~~ public class ReceiveLogs2Console { private static final String EXCHANGE_NAME = "topic_logs"; public static void main(String[] argv) throws IOException, InterruptedException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("127.0.0.1"); // 打開連接和創建頻道，與發送端一樣 Connection connection = factory.newConnection(); final Channel channel = connection.createChannel(); //聲明exchange channel.exchangeDeclare(EXCHANGE_NAME, "topic"); // 聲明一個隨機隊列 String queueName = channel.queueDeclare().getQueue(); String[] routingKeys ={"auth.*","*.info","#.warning"};//關注所有的授權日志、所有info和waring級別的日志 for (String routingKey : routingKeys) { //關注所有級別的日志（多重綁定） channel.queueBind(queueName, EXCHANGE_NAME, routingKey); } System.out.println(" [*] Waiting for messages. To exit press CTRL+C"); // 創建隊列消費者 final Consumer consumer = new DefaultConsumer(channel) { @Override public 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 [" + envelope.getRoutingKey() + "] :'" + message + "'"); } }; channel.basicConsume(queueName, true, consumer); } } ~~~ 消費者2（ReceiveLogs2File.java） ~~~ public class ReceiveLogs2File { private static final String EXCHANGE_NAME = "topic_logs"; public static void main(String[] argv) throws IOException, InterruptedException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("127.0.0.1"); // 打開連接和創建頻道，與發送端一樣 Connection connection = factory.newConnection(); final Channel channel = connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, "topic"); // 聲明一個隨機隊列 String queueName = channel.queueDeclare().getQueue(); channel.queueBind(queueName, EXCHANGE_NAME, ""); String severity="kern.error";//只關注核心錯誤級別的日志，然后記錄到文件中去。 channel.queueBind(queueName, EXCHANGE_NAME, severity); System.out.println(" [*] Waiting for messages. To exit press CTRL+C"); // 創建隊列消費者 final Consumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String message = new String(body, "UTF-8"); //記錄日志到文件： print2File( "["+ envelope.getRoutingKey() + "] "+message); } }; channel.basicConsume(queueName, true, consumer); } private static void print2File(String msg) { try { String dir = ReceiveLogs2File.class.getClassLoader().getResource("").getPath(); String logFileName = new SimpleDateFormat("yyyy-MM-dd").format(new Date()); File file = new File(dir, logFileName + ".log"); FileOutputStream fos = new FileOutputStream(file, true); fos.write((new SimpleDateFormat("HH:mm:ss").format(new Date())+" - "+msg + "\r\n").getBytes()); fos.flush(); fos.close(); } catch (FileNotFoundException e) { e.printStackTrace(); } catch (IOException e) { e.printStackTrace(); } } } ~~~ ? ? ?最終結果：