當前位置：首頁 > 编程语言 > java >内容正文

java

(RabbitMQ) Java Client API Guide

發(fā)布時間：2024/4/11 java 26 豆豆

生活随笔收集整理的這篇文章主要介紹了 (RabbitMQ) Java Client API Guide 小編覺得挺不錯的,現(xiàn)在分享給大家,幫大家做個參考.

歡迎支持筆者新作：《深入理解Kafka:核心設計與實踐原理》和《RabbitMQ實戰(zhàn)指南》，同時歡迎關注筆者的微信公眾號：朱小廝的博客。

本篇翻譯的是RabbitMQ官方文檔關于API的內容，原文鏈接：http://www.rabbitmq.com/api-guide.html。博主對其內容進行大體上的翻譯，有些許部分會保留英文，個人覺得這樣更加有韻味，如果全部翻譯成中文，會存在偏差，文不達意（主要是功力淺薄~~）。文章也對部分內容進行一定的解釋，增強對相關知識點的理解。

Overview

RabbitMQ java client uses com.rabbitmq.client as its top-level package, 關鍵的classes和interface如下：

Channel
Connection
ConnectionFactory
Consumer

AMQP協(xié)議層面的操作通過Channel接口實現(xiàn)。Connection是用來open Channels的，可以注冊event handlers，也可以在結束是close connections. Connection是通過ConnectionFactory來進行初始化操作的，當然也需要配置不同的connection設置，比如vhost或者username等。

Connections and Channels

關鍵的API如Connection和Channel，分別代表了AMQP-0-9-1的connection和channel。典型的包導入如下：

import com.rabbitmq.client.Connection; import com.rabbitmq.client.Channel;

Connecting to a broker

下面的代碼用來在給定的參數(shù)（hostname, port number等）下連接一個AMQP broker:

ConnectionFactory factory = new ConnectionFactory(); factory.setUsername(userName); factory.setPassword(password); factory.setVirtualHost(virtualHost); factory.setHost(hostName); factory.setPort(portNumber); Connection conn = factory.newConnection();

也可以選擇使用URI來實現(xiàn)，示例如下：

ConnectionFactory factory = new ConnectionFactory(); factory.setUri("amqp://userName:password@hostName:portNumber/virtualHost"); Connection conn = factory.newConnection();

Connection接口被用來open一個channel:

Channel channel = conn.createChannel();

這樣在創(chuàng)建之后，Channel可以用來發(fā)送或者接受消息了。
在使用完之后，關閉連接：

channel.close(); conn.close();

顯示的關閉channel是一個很好的習慣，但這不是必須的，在基本的connection關閉的時候channel也會自動的關閉。

Using Exchanges and Queues

AMQP的high-level構建模塊exchanges和queues是Client端應用所必須的。在使用之前必須先“declared”(聲明)，確保在使用之前已經(jīng)存在，如果不存在則創(chuàng)建它，這些操作都包含在declare里。
下面的代碼是演示如何declare一個exchange和queue：

channel.exchangeDeclare(exchangeName, "direct", true); String queueName = channel.queueDeclare().getQueue(); channel.queueBind(queueName, exchangeName, routingKey);

上面創(chuàng)建了一個durable, non-autodelete并且綁定類型為direct的exchange以及一個non-durable, exclusive,autodelete的queue（此queue的名稱由broker端自動生成）。這里的exchange和queue也都沒有設置特殊的arguments。

上面的代碼也展示了如果使用routing key將queue和exchange綁定起來。上面聲明的queue具備如下特性：排他的（只對當前client同一個Connection可用, 同一個Connection的不同的Channel可共用），并且也會在client連接斷開時自動刪除。

如果要在client共享一個queue，可以做如下聲明：

channel.exchangeDeclare(exchangeName, "direct", true); channel.queueDeclare(queueName, true, false, false, null); channel.queueBind(queueName, exchangeName, routingKey);

這里的queue是durable的，非排他的，non-autodelete, 而且也有一個確定的已知的名稱（又Client指定而非broker端自動生成）。

注意：Channel的API方法都是可以重載的，比如exchangeDeclare，queueDeclare根據(jù)參數(shù)的不同，可以有不同的重載形式，根據(jù)自身的需要去進行調用。

Publish messages

如果要發(fā)送一個消息可以采用Channel.basicPublish的方式：

byte[] messageBodyBytes = "Hello, world!".getBytes(); channel.basicPublish(exchangeName, routingKey, null, messageBodyBytes);

為了更好的控制，你也可以使用mandatory這個屬性，或者可以發(fā)送一些特定屬性的消息：

channel.basicPublish(exchangeName, routingKey, mandatory,MessageProperties.PERSISTENT_TEXT_PLAIN,messageBodyBytes);

這個方法發(fā)送了一條消息，這條消息的delivery mode為2，即消息需要被持久化在broker中，同時priority優(yōu)先級為1，content-type為text/plain。你可以可以自己設定消息的屬性：

channel.basicPublish(exchangeName, routingKey,new AMQP.BasicProperties.Builder().contentType("text/plain").deliveryMode(2).priority(1).userId("bob").build()),messageBodyBytes);

你也可以發(fā)送一條帶有header的消息：

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);

你也可以發(fā)送一條帶有超時時間expiration的消息：

channel.basicPublish(exchangeName, routingKey,new AMQP.BasicProperties.Builder().expiration("60000").build()),messageBodyBytes);

以上只是舉例，由于篇幅關系，這里就不一一列舉所有的可能情形了。

Channel#basicPublish方法在以下兩種情形下會被阻塞，具體可以參考http://www.rabbitmq.com/alarms.html：

When memory use goes above the configured limit.（內存不夠）
When disk space drops below the configured limit.（磁盤空間不足）

Channles and Concurrency Consideration(Thread Safaty)

Channel實例不能在線程建共享，應用程序應該為每一個線程開辟一個Channel, 而不是在多線程建共享Channel。某些情況下Channel的操作可以并發(fā)運行，但是某些情況下并發(fā)會導致在網(wǎng)絡上錯誤的幀交叉，同時也會影響publisher confirm, 故多線程共享Channel是非線程安全的。

Receiving messages by subscription

import com.rabbitmq.client.Consumer; import com.rabbitmq.client.DefaultConsumer;

接受消息一般是通過實現(xiàn)Consumer接口或者繼承DefaultConsumer來實現(xiàn)。當調用與Consumer相關的API方法時，不同的訂閱采用consumer tags以作彼此的區(qū)分，在同一個Channel中的Consumer也需要通過唯一的consumer tags以作區(qū)分。。

消費消息demo如下：

boolean autoAck = false; channel.basicConsume(queueName, autoAck, "myConsumerTag",new DefaultConsumer(channel) {@Overridepublic 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, 對于Consumer來說這個設置是非常必要的。（譯者注：具體可以參考RabbitMQ之消息確認機制（事務+Confirm）中Consumer確認那一章節(jié)。）

同時對于Consumer來說重寫handleDelivery方法也是十分方便的。更復雜的Consumer會重寫（override）更多的方法，比如handleShutdownSignal當channels和connections close的時候會調用，handleConsumeOk在其他callback方法之前調用，返回consumer tags.

Consumer同樣可以override handleCancelOk和handleCancel方法，這樣在顯示的或者隱式的取消的時候調用。

你可以通過Channel.basicCancel方法顯示的cancel一個指定的Consumer:

channel.basicCancel(consumerTag);

(譯者注：這句代碼首先觸發(fā)handleConsumerOk，之后觸發(fā)handleDelivery方法，最后觸發(fā)handleCancelOk方法。)

單個Consumer在Connection上都分配單個的線程來調用這些callback的方法，也就是說Consumer這里安全的調用阻塞式的方法，比如queueDeclare, txCommit, basicCancel或者basicPublish。

每個Channel都有自己的獨立的線程。最常用的用法是一個Channel對應一個Consumer, 也就是意味著Consumers彼此間沒有任何關聯(lián)。當然你也可以在一個Channel中維持多個Consumers, 但是要注意一個問題，如果在Channel的一個Consumer一直在運行，那么對于其他Consumer的callbacks而言會被hold up(耽擱)。

Retrieving individual messages

通過Channel.basicGet可以一個一個的獲取消息，其返回值是GetResponse（from which the header information(properties) and message body can be extracted）。
示例Demo如下：

boolean autoAck = false; GetResponse response = channel.basicGet(queueName, autoAck); if (response == null) {// No message retrieved. } else {AMQP.BasicProperties props = response.getProps();byte[] body = response.getBody();long deliveryTag = response.getEnvelope().getDeliveryTag();...

如果設置autoAck為false，那么你同樣需要顯示的調用Channel.basicAck來確認消息已經(jīng)被成功的接受了：

channel.basicAck(method.deliveryTag, false); // acknowledge receipt of the message

(譯者注：有關RabbitMQ的消費端的更多信息可以參考：RabbitMQ之Consumer消費模式（Push & Pull）)

Handing unroutable messages

如果一個消息在publish的時候設置了mandatory標記，如果消息沒有成功的路由到某個隊列的時候，broker端會通過Basic.Return返回回來。

這時候客戶端需要實現(xiàn)ReturnListener這個接口，并且調用Channel.setReturnListener。如果client沒有配置相關的return listener那么相應的需要被returned的消息就會被drop掉。

channel.setReturnListener(new ReturnListener() {public void handleBasicReturn(int replyCode,String replyText,String exchange,String routingKey,AMQP.BasicProperties properties,byte[] body)throws IOException {...} });

（譯者注：有關mandatory的更多內容可以參考：RabbitMQ之mandatory和immediate。）

Shutdown Protocol

Overview of the AMQP client shutdown

AMQP-0-9-1的connection和channel采用同樣的方式來管理網(wǎng)絡失敗，內部錯誤以及顯示的local shutdown。

AMQP-0-9-1的connection和channel具備如下的生命周期狀態(tài)（lifecycle states）:

open: the object is ready to use.
closing：當前對象被顯示的通知調用shutdown，這樣就產(chǎn)生了一個shutdown的請求至lower-layer的對象進行相應的操作，并等待這些shutdown操作的完成。
closed：當前對象已經(jīng)接受到所有的shutdown完成的通知，并且也shutdown了自身。
這些對象最終成closed的狀態(tài)，而不管是由于什么原因引起的，或者是一個applicatin request，或者是內部client library的失敗，或者是a remote network request，亦或者是network failure。

AMQP的connecton和channel對象控制（possess）了shutdown-related的方法：addShutdownListener(ShutdownListener listener)和removeShutdownListener(ShutdownListener listener)。當connection和channel轉向closed狀態(tài)時會調用ShutdownListener, 而且如果將一個ShutdownListener注冊到一個已經(jīng)處于closed狀態(tài)的object(特指connection或者channel的對象)時，會立刻調用ShutdownListener。

getCloseReason():可以讓你知道the object’s shutdown的原因。
isOpen()：檢測the objects當前的是否處于open狀態(tài)。
close(int closeCode, String closeMessage)：顯示的通知the object執(zhí)行shutdown。

示例代碼：

import com.rabbitmq.client.ShutdownSignalException; import com.rabbitmq.client.ShutdownListener;connection.addShutdownListener(new ShutdownListener() {public void shutdownCompleted(ShutdownSignalException cause){...} });

Information about the ircumstances of a shutdown

當觸發(fā)ShutdownListener的時候，就可以獲取到ShutdownSignalException，這個ShutdownSignalException包含了close的原因，這個原因也可以通過getCloseReason()方法獲取。

ShutdownSignalException提供了多個方法用來分析shutdown的原因。isHardError()方法可以知道是connection還是channel的error，getReason()方法可以獲取cause相關的信息（以AMQP method的形式，com.rabbitmq.client.Method：AMQP.Channel.Close or AMQP.Connection.Close）:

public void shutdownCompleted(ShutdownSignalException cause) {if (cause.isHardError()){Connection conn = (Connection)cause.getReference();if (!cause.isInitiatedByApplication()){Method reason = cause.getReason();...}...} else {Channel ch = (Channel)cause.getReference();...} }

Atomicity and use of the isOpen() method

我們并不推薦在生產(chǎn)環(huán)境的代碼上使用channel或者connection的isOpen()方法，這個isOpen()方法的返回值依賴于shutdown cause的存在，有可能會產(chǎn)生競爭。
（譯者添加：關于isOpen依賴于shutdown cause, isOpen的實現(xiàn)代碼如下：）

public boolean isOpen() {synchronized(this.monitor) {return this.shutdownCause == null;}}

錯誤的使用方式如下：

public void brokenMethod(Channel channel) {if (channel.isOpen()){// The following code depends on the channel being in open state.// However there is a possibility of the change in the channel state// between isOpen() and basicQos(1) call...channel.basicQos(1);} }

正確的使用方式：

public void validMethod(Channel channel) {try {...channel.basicQos(1);} catch (ShutdownSignalException sse) {// possibly check if channel was closed// by the time we started action and reasons for// closing it...} catch (IOException ioe) {// check why connection was closed...} }

Advanced Connection options

Consumer thread pool

默認情況下客戶端會自動分配一個ExecutorService給Consumer線程，同樣你也可以使用自定義的線程池，比如：

ExecutorService es = Executors.newFixedThreadPool(20); Connection conn = factory.newConnection(es);

當connection關閉的時候，默認的ExecutorService會被shutdown,但是如果是自定義的ExecutorService將不會被自動的shutdown，所以Clients程序需要在最終關閉的時候手動的去執(zhí)行shutdown()，否則將會阻止JVM的正常關閉。

同一個executor service可以被多個connections共用。除非有明顯的證據(jù)證明默認的ExecutorService不能滿足當前Consumer callbacks的需要，否則不建議使用自定義的ExecutorService.

Using Lists of Hosts

可以通過使用Address來執(zhí)行newConnection(). com.rabbitmq.client.Address的使用是比較方便的，例如：

Address[] addrArr = new Address[]{ new Address(hostname1, portnumber1), new Address(hostname2, portnumber2)}; Connection conn = factory.newConnection(addrArr);

如果hostname1:portnumber1成功了連接，而hostname2:portnumber2連接失敗了，connection照樣會成功returned, 也不會跑出IOException。這個和你重復設置host，port然后調用factory.newConnection()直到有一組成功為止一個效果。

同樣可以指定自定義的ExecutorService, 比如：factory.newConnection(es, addrArr)。

If you want moew control over the host to connect to, see the support for service discovery.

Service discovery with the AddressResolver interface

在版本3.6.6開始，可以通過AddressResolver接口的實現(xiàn)來創(chuàng)建connection:

Connection conn = factory.newConnection(addressResolver);

AddressResolver接口如下：

public interface AddressResolver {List<Address> getAddresses() throws IOException; }

使用AddressResolver可以更好的實現(xiàn)custom service discovery邏輯，和“automatic recovery”組合使用，客戶端可以自動的和broker nodes連接.AddressResolver也可以有效的配合負載均衡策略。

AddressResolver有兩個實現(xiàn)：DnsRecordIpAddressResolver和DnsSrvRecordAddressResolver.(博主沒用過AddressResolver，這里就不多做解釋了)

Heartbeat Timeout

有關Heartbeat的內容請參考Heatbeats guide。（原文就是這么說的。）

Custom Thread Factories

略。和Google App Engine有關。

Support for Java non-blocking IO

4.0版本開始客戶端引入了java的NIO，這里引入NIO的目的不是為了比BIO的更快，而是是的更加容易的控制資源。

對于默認的BIO模式，每個connection都需要一個獨立的線程來進行網(wǎng)絡通訊。但在NIO模式下，你可以控制網(wǎng)絡通訊讀寫線程的數(shù)量。

如果你的java程序需要許多的connections(幾十個或者幾百個)，那么使用NIO模式是一個很好的選擇。相比BIO而言，你所使用的線程數(shù)很少，通過設置合理的線程數(shù)，你可以不必擔心性能的損耗，尤其是在connections不怎么busy的時候。

NIO必須被顯示的設置：

ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.useNio();

你也可以設置NIO的參數(shù)：

connectionFactory.setNioParams(new NioParams().setNbIoThreads(4));

NIO模式下使用合理的默認值，同時你也可以根據(jù)自身的負載情況來進行合理的變換。

Automatic Recovery From Network Failures

Connection Recovery

客戶端和broker之間的網(wǎng)絡通訊可能會失敗。RabbitMQ java client支持connections和拓撲topology(指queues, exchanges, bindings and consumers)的自動回復。自動恢復過程有如下幾個步驟：

Reconnect
Restore connection listeners
Re-open channels
Restore channel listeners
Restore channel basic.qos setting, publisher confirms and transaction settings

topology的恢復包括如下行為，performed for every channel:

Re-declare exchange (exception for predefined ones)
Re-declare queues
Recover all bindings
Recover all consumers

在版本4.0.0開始，自動回復默認是開啟的。你也通過factory.setAutomaticRecoveryEnabled(boolean)可以手動的設置automatic onnection recovery.

如果由于某些異常（比如RabbitMQ節(jié)點始終連接不上）而導致的恢復失敗。那么會在某個特定的時間間隔內重試，默認此間隔為5s，當然此值可配：

ConnectionFactory factory = new ConnectionFactory(); // attempt recovery every 10 seconds factory.setNetworkRecoveryInterval(10000);

Recovery Listeners

It is possible to register one or more recovery listeners on recoverable connections and channels. 當connection recovery啟用的時候，通過調用ConnectionFactory#newConnection和Connection#createChannel返回的connections實現(xiàn)com.rabbitmq.client.Recoverable. 這里提供了兩個方法：addRecoveryListener和removerRecoveryListener.

/*** Provides a way to register (network, AMQP 0-9-1) connection recovery* callbacks.** When connection recovery is enabled via {@link ConnectionFactory},* {@link ConnectionFactory#newConnection()} and {@link Connection#createChannel()}* return {@link Recoverable} connections and channels.** @see com.rabbitmq.client.impl.recovery.AutorecoveringConnection* @see com.rabbitmq.client.impl.recovery.AutorecoveringChannel*/ public interface Recoverable {/*** Registers a connection recovery callback.** @param listener Callback function*/void addRecoveryListener(RecoveryListener listener);void removeRecoveryListener(RecoveryListener listener); }

當然你必須將connections和channels強制轉換為Recoverable的才能使用這些方法。

Effects on Publishing

消息通過Channel,basicPublish發(fā)布，如果connection down了那么消息就會丟失。客戶端不會在connection恢復之后重新delivery這些消息。為了確保消息的可靠性，可以參考Publisher Confims.(或者可以參考博主的博文：RabbitMQ之消息確認機制（事務+Confirm）)。

Topology Recovery

Topology recovery涉及到exchanges, queues, bindings and consumer.當automatic recovery可用時topology recovery默認也可用。當然topology也可顯示的設置為disabled：

ConnectionFactory factory = new ConnectionFactory();Connection conn = factory.newConnection(); // enable automatic recovery (e.g. Java client prior 4.0.0) factory.setAutomaticRecoveryEnabled(true); // disable topology recovery factory.setTopologyRecoveryEnabled(false);

Manual Acknowledgements and Automatic Recovery

當autoAck設置為false的時候，在消息delivery和ack的時候有可能會由于網(wǎng)絡原因故障，在connection recovery之后，RabbitMQ會將所有的channels的delivery tags進行重置。這就意味著basic.ack, basic,nack以及basic.reject帶有old delivery tags的將會引起channel exception。為了解決這個為題，RabbitMQ java client會記錄和更新相應的delivery tags來確保在恢復期間保持單調遞增。帶有過時的delivery tags的ack將不會被發(fā)送。采用manual ack和automatic recovery的應用必須具備處理redeliveries的能力。

Unhandled Exceptions

在connection, channel, recovery, consumer生命周期內涉及的未被處理的異常可以委托給exception handler. Exception handler實現(xiàn)了ExceptionHandler這個接口，默認情況下使用的是DefaultExceptionHandler, 只是在標準輸出流中打印一些exception的細節(jié)。

你可以使用ConnectionFactory#setExceptionHandler來override這個handler,這個handler可以被ConnectionFactory創(chuàng)建的所有的Connections所使用：

ConnectionFactory factory = new ConnectionFactory(); factory.setExceptionHandler(customHandler);

Metrics and monitoring

RabbitMQ Java Client on Google App Engine

Cavets and Limitations

The RPC Pattern