[TOC] # 簡介 首先加入RabbitMQ java client依賴： ~~~xml <dependencies> <dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>4.0.2</version> </dependency> </dependencies> ~~~ RabbitMQ的java client使用`com.rabbitmq.client`作為其頂級包。關鍵的類和接口是： ~~~ com.rabbitmq.client.Channel com.rabbitmq.client.Connection com.rabbitmq.client.ConnectionFactory com.rabbitmq.client.Consumer ~~~ 通過Channel可以進行一系列的api操作。 Connection（連接）用于打開通道，注冊連接生命周期事件處理程序，并關閉不再需要的連接。 Connection（連接）通過ConnectionFactory實例化，ConnectionFactory可以設置一些Collection（連接）的一些配置，比如說vhost或者說username等等。 ## Connections（連接）和Channels（管道） 核心的類是Connections（連接）和Channels（管道），分別代表著AMQP 0-9-1協議中的Connections（連接）和Channels（管道），一般被導入 ~~~ import com.rabbitmq.client.Connection; import com.rabbitmq.client.Channel; ~~~ ## 連接服務器 下面的代碼時使用給定的參數（host name，端口等等）連接AMQP的服務器。 ~~~ ConnectionFactory factory = new ConnectionFactory(); factory.setUsername(userName); factory.setPassword(password); factory.setVirtualHost(virtualHost); factory.setHost(hostName); factory.setPort(portNumber); Connection conn = factory.newConnection(); ~~~ 所有的這些參數RabbitMQ服務器都設置了默認值，可以在ConnectionFactory類中查看這些默認值。 另外，URI可以以下面的方法進行連接都有默認值。 ~~~ ConnectionFactory factory = new ConnectionFactory(); factory.setUri("amqp://userName:password@hostName:portNumber/virtualHost"); Connection conn = factory.newConnection(); ~~~ Connection（連接）接口可以被用作創建一個channel（管道）： ~~~ Channel channel = conn.createChannel(); ~~~ 可以使用channel（管道）發送和接收消息，下面會有講到。 關閉連接，只需要關閉channel（管道）和connection（連接）： ~~~ channel.close(); conn.close(); ~~~ 注意，關閉管道是被認為是最佳實踐，但是卻不是嚴格意義的必要的。當底層的連接關閉時候，channel（管道）也就自動的被關閉了。 ## 使用Exchanges和Queues 客戶端應用必須應用在exchanges和queues，這些都是AMQP協議定義的。使用這些（exchanges和queues）首先必須“聲明”它（就是創建的意思）。 下面的代碼就是怎樣去"聲明"一個exchange和隊列，并且將它們綁定在一起。 ~~~ channel.exchangeDeclare(exchangeName, "direct", true); String queueName = channel.queueDeclare().getQueue(); channel.queueBind(queueName, exchangeName, routingKey); ~~~ 可以通過參數去設置exchange和queue的一些屬性，使用這些方法的一些重載方法進行相關設置。 ~~~ channel.exchangeDeclare(exchangeName, "direct", true); channel.queueDeclare(queueName, true, false, false, null); channel.queueBind(queueName, exchangeName, routingKey); ~~~ ## 發送消息（Publishing messages） 使用Channel.basicPublish方法將消息發送給一個exchange： ~~~ byte[] messageBodyBytes = "Hello, world!".getBytes(); channel.basicPublish(exchangeName, routingKey, null, messageBodyBytes); ~~~ 為了更好的控制，你可以使用重載的參數來設置消息的一些屬性（比如說mandatory標志，關于mandatory標志，下面會講到），或者在發送消息前設定一些消息屬性。 ~~~ channel.basicPublish(exchangeName, routingKey, mandatory, MessageProperties.PERSISTENT_TEXT_PLAIN, messageBodyBytes); ~~~ 可以自己構建BasicProperties的對象，如下面的代碼： ~~~ channel.basicPublish(exchangeName, routingKey, new AMQP.BasicProperties.Builder() .contentType("text/plain") .deliveryMode(2) .priority(1) .userId("bob") .build()), messageBodyBytes); ~~~ 發送消息指定頭信息： ~~~ Map<String, Object> headers = new HashMap<String, Object>(); headers.put("latitude", 51.5252949); headers.put("longitude", -0.0905493); channel.basicPublish(exchangeName, routingKey, new AMQP.BasicProperties.Builder() .headers(headers) .build()), messageBodyBytes); ~~~ 發送一個有過期時間的消息，下面的博客也會講到： ~~~ channel.basicPublish(exchangeName, routingKey, new AMQP.BasicProperties.Builder() .expiration("60000") .build()), messageBodyBytes); ~~~ ## 訂閱消息（"Push API"） ~~~ import com.rabbitmq.client.Consumer; import com.rabbitmq.client.DefaultConsumer; ~~~ 最有效的接收消息的方法是使用Consumer接口去訂閱。當消息到達消費端的時候會自動的傳遞消費（delivered），而不需要去請求。 當我們調用Consumers（消費者）有關的api的時候，會生成一個消費者標識符（consumer tag）。 不同的Consumer實例必須有不同的消費者標簽。 強烈建議不要在連接上重復使用消費者標簽，不然在監視消費者時可能導致自動連接恢復和混淆監控數據的問題。 實現Consumer的最簡單的方法是將便利（convenience）類DefaultConsumer子類化。 該子類的對象可以在basicConsume方法調用中傳遞以設置訂閱： ~~~java boolean autoAck = false; channel.basicConsume(queueName, autoAck, "myConsumerTag", new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String routingKey = envelope.getRoutingKey(); String contentType = properties.getContentType(); long deliveryTag = envelope.getDeliveryTag(); // (process the message components here ...) channel.basicAck(deliveryTag, false); } }); ~~~ 在這里，因為我們設置了自動確認（`autoAck`）的值為false，所以有必要在傳遞給消費者的方法中進行自動確認（`handleDelivery`方法中）。 更復雜的消費者將會重寫更多的方法。事實上，`handleShutdownSignal`方法被調用當Channel（通道）和連接關閉的時候。并且在調用該消費者的任何回調方法之前將`consumer tag`傳遞給`handleConsumeOk`（com.rabbitmq.client.Consumer接口中定義的方法）方法 消費者還可以分別實現`handleCancelOk`（com.rabbitmq.client.Consumer接口中定義的方法）和`handleCancel`（com.rabbitmq.client.Consumer接口中定義的方法）方法來通知顯式和隱式取消。 你也可以使用Channel.basicCancel方法明確的取消一個特定的消費，傳遞consumer tag， ~~~ channel.basicCancel(consumerTag); ~~~ 和生產者一樣，對于消費者來說并發處理消息也要慎重考慮。 回調給消費者是在與實例化其`Channel`（管道）的線程分開的線程池中調度的。 這意味著消費者可以安全地在`Connection`或`Channel`上調用阻塞方法，例如`Channel＃queueDeclare`或`Channel＃basicCancel`。 每一個`Channel`（管道）都有自己的調度線程。對于最常用的使用方式就是一個消費者一個`Channel`（管道），意味著一個消費者不會阻塞其他的消費。如果是一個`Channel`（管道）多消費者必須明白一個長時間的消費調用可能會阻塞其他消費者的回調調度。 # 通道和并發注意事項（線程安全） 根據經驗，**在線程間共享Channel（通道）是要避免的**。應用應該優先使用每個線程自己的Channel（通道）實例，而不是多個線程共享這個Channel（通道）實例。 雖然有些在Channel（通道）上的操作是可以并發安全的調用，但是一些操作不行會導致一些邊界交錯，雙重確認等等。 在共享（多線程）Channel（通道）上進行并發發布會導致一些邊界交錯，觸發連接協議異常和連接關閉。因此需要嚴格在應用中同步調用（Channel#basicPublish必須在正確關鍵的地方調用）。線程之間的共享也會干擾生產者的消息確認。我們強烈的推薦不應該在通道上進行并發的發布消息。 在共享的Channel（通道）上一個線程生產（publish）消息，一個線程消費（consume）消息是線程安全的。 服務器推送可以同時發送，保證每通道的訂閱被保留。 調度機制使用`java.util.concurrent.ExecutorService`。 可以使用單列的`ConnectionFactory`調用`ConnectionFactory#setSharedExecutor`去設置所有連接共用的`executor`。 當我們手動確認[manual acknowledgements](https://link.jianshu.com?t=http://www.rabbitmq.com/confirms.html) 的時候，很重要的是考慮什么線程去做這個ack確認。如果接收傳遞的線程（例如，Consumer＃handleDelivery委托給不同線程的傳遞處理）不同于手動確認的線程，則將多個線程參數設置為true是線程不安全的并導致雙重確認，因此導致通道協議異常導致Channel關閉。一次確認一條消息可以確保安全的。 # 簡單例子 通過ConnectionFactory獲得Connection，Connection得到Channel **Exchange** ~~~ package rabbitmq; import com.rabbitmq.client.*; import java.io.IOException; import java.util.HashMap; import java.util.Map; import java.util.concurrent.TimeoutException; public class ExchangeTest { public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setHost("127.0.0.1"); connectionFactory.setPort(5672); connectionFactory.setUsername("guest"); connectionFactory.setPassword("guest"); connectionFactory.setVirtualHost("/"); Connection connection = connectionFactory.newConnection(); Channel channel = connection.createChannel(); //創建exchange,類型是direct類型 channel.exchangeDeclare("zhihao.miao", "direct"); //創建exchange, 類型是direct類型 channel.exchangeDeclare("zhihao.miao.info", BuiltinExchangeType.DIRECT); //第三個參數表示是否持久化,同步操作,有返回值 AMQP.Exchange.DeclareOk ok = channel.exchangeDeclare("zhihao.miao.debug", BuiltinExchangeType.DIRECT, true); System.out.println("---ok---" + ok); //設置屬性 Map<String, Object> argument = new HashMap<>(); argument.put("alternate-exchange", "log"); //第三個是持久化,第四個是是否自動刪除 channel.exchangeDeclare("zhihao.miao.warn", BuiltinExchangeType.TOPIC, true, false, argument); //異步創建exchange，沒有返回值, channel.exchangeDeclareNoWait("zhihao.miao.log", BuiltinExchangeType.TOPIC, true, false, false, argument); //判斷是否存在, 不存在就報錯 AMQP.Exchange.DeclareOk declareOk = channel.exchangeDeclarePassive("zhihao.miao.info"); System.out.println("---declareOk---" + declareOk); //判斷是否存在, 不存在就報錯 declareOk = channel.exchangeDeclarePassive("zhihao.miao.debug"); System.out.println("---declareOk2---" + declareOk); //刪除exchange(可重復執行), 刪除一個不存在的也不會報錯 channel.exchangeDelete("zhihao.miao"); channel.exchangeDelete("zhihao.miao.debug"); channel.exchangeDelete("zhihao.miao.info"); channel.exchangeDelete("zhihao.miao.warn"); channel.exchangeDelete("zhihao.miao.log"); //刪除exchange channel.exchangeDelete("zhihao.miao.info"); channel.close(); connection.close(); } } ~~~ 隊列的api操作 **queues** ~~~ public class QueueTest { public static void main(String[] args) throws Exception{ ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setHost("192.168.1.131"); connectionFactory.setPort(5672); connectionFactory.setUsername("zhihao.miao"); connectionFactory.setPassword("123456"); connectionFactory.setVirtualHost("/"); Connection connection = connectionFactory.newConnection(); Channel channel = connection.createChannel(); //第二個參數表示是否持久化，第三個參數是判斷這個隊列是否在連接是否生效，為true表示連接關閉隊列刪除。 AMQP.Queue.DeclareOk ok = channel.queueDeclare("zhihao.info",true,false,false,null); System.out.println(ok); //異步沒有返回值的方法api channel.queueDeclareNoWait("zhihao.info.miao",true,false,false,null); //判斷queue是否存在，不存在會拋出異常 //channel.exchangeDeclarePassive("zhihao.info"); //拋出錯誤 //channel.exchangeDeclarePassive("zhihao.info.miao2"); //exchange和queue進行綁定（可重復執行，不會重復創建） channel.queueBind("zhihao.info","zhihao.miao.order","info"); //異步進行綁定 channel.queueBindNoWait("zhihao.info.miao","zhihao.miao.pay","info",null); //exchange與exchange進行綁定(可重復執行，不會重復創建） channel.exchangeBind("zhihao.miao.email","zhihao.miao.weixin","debug"); //exchange和queue進行解綁（可重復執行） channel.queueUnbind("zhihao.info","zhihao.miao.order","info"); //exchange和exchange進行解綁（可重復執行） channel.exchangeUnbind("zhihao.info.miao","zhihao.miao.pay","debug"); //刪除隊列 channel.queueDelete("zhihao.info"); channel.close(); connection.close(); } } ~~~ **消息的發送** ~~~java public class Sender { public static void main(String[] args) throws Exception{ ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setUri("amqp://zhihao.miao:123456@192.168.1.131:5672"); Connection connection = connectionFactory.newConnection(); Channel channel = connection.createChannel(); AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder().deliveryMode(2). contentEncoding("UTF-8").build(); //第一個參數是exchange參數，如果是為空字符串，那么就會發送到(AMQP default)默認的exchange，而且routingKey //便是所要發送到的隊列名 channel.basicPublish("","zhihao.info.miao",properties,"忘記密碼，驗證碼是1234".getBytes()); channel.basicPublish("","zhihao.miao",properties,"忘記密碼，六位驗證密碼是343sdf".getBytes()); //direct類型的exchange類型的exchange，zhihao.miao.order綁定zhihao.info.miao隊列，route key是order channel.basicPublish("zhihao.miao.order","order",properties,"愛奇藝會員到期了".getBytes()); //zhihao.miao.pay綁定zhihao.info.miao隊列，route key是order channel.basicPublish("zhihao.miao.pay","pay",properties,"優酷會員到期了".getBytes()); //topic類型的exchange channel.basicPublish("log","user.log",properties,"你的外賣已經送達".getBytes()); channel.basicPublish("log","user.log.info",properties,"你的外賣正在配送中".getBytes()); channel.basicPublish("log","user",properties,"你的投訴已經采納".getBytes()); channel.close(); connection.close(); } } ~~~ **消息消費** ~~~ public class Consumer { public static void main(String[] args) throws Exception{ ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setHost("192.168.1.131"); connectionFactory.setPort(5672); connectionFactory.setUsername("zhihao.miao"); connectionFactory.setPassword("123456"); connectionFactory.setVirtualHost("/"); //客戶端的消費消息 Map<String,Object> clientProperties = new HashMap<>(); clientProperties.put("desc","支付系統2.0"); clientProperties.put("author","zhihao.miao"); clientProperties.put("user","zhihao.miao@xxx.com"); connectionFactory.setClientProperties(clientProperties); //給客戶端的connetction命名 Connection connection = connectionFactory.newConnection("log隊列的消費者"); //給channel起個編號 Channel channel = connection.createChannel(10); //返回consumerTag，也可以通過重載方法進行設置consumerTag String consumerTag = channel.basicConsume("user_log_queue",true,new SimpleConsumer(channel)); System.out.println(consumerTag); TimeUnit.SECONDS.sleep(30); channel.close(); connection.close(); } } ~~~ 具體的消息邏輯，繼承DefaultConsumer類重寫handleDelivery方法，如果是手工確認消息，會在handleDelivery方法中進行相關的確認（調用相關api） ~~~java public class SimpleConsumer extends DefaultConsumer{ public SimpleConsumer(Channel channel){ super(channel); } @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(consumerTag); System.out.println("-----收到消息了---------------"); System.out.println("消息屬性為："+properties); System.out.println("消息內容為："+new String(body)); } } ~~~