要在Java中獲得高性能RabbitMQ消息傳遞，您必須優(yōu)化客戶(hù)端和經(jīng)紀(jì)人配置。 1。使用帶有緩存的通道池的CachingConnectionFactory使用連接和通道池來(lái)減少開(kāi)銷(xiāo)。 2。啟用發(fā)布者會(huì)異步確認(rèn)，並使用批處理確認(rèn)或帶有輸送標(biāo)籤的異步跟蹤，以進(jìn)行可靠的高速發(fā)布。 3。通過(guò)使用有效的序列化（如Protobuf）來(lái)最大程度地減少消息大小，並在有益的情況下用GZIP壓縮大量有效載荷。 4。設(shè)置使用基本QO的消費(fèi)者預(yù)取計(jì)數(shù)以平衡負(fù)載並防止緩慢的消費(fèi)者阻止其他人。 5。在單個(gè)連接上使用多個(gè)通道進(jìn)行單獨(dú)的發(fā)布和消費(fèi)任務(wù)，以避免線(xiàn)條阻止。 6。通過(guò)使用SSD，監(jiān)視隊(duì)列長(zhǎng)度，啟用懶惰或Quorum隊(duì)列以及使用鏡像隊(duì)列以高可用性來(lái)調(diào)整RabbitMQ經(jīng)紀(jì)人設(shè)置。 7.通過(guò)重新連接邏輯，掌握消費(fèi)者和斷路器來(lái)實(shí)現(xiàn)強(qiáng)大的故障處理，以保持負(fù)載下的彈性。通過(guò)結(jié)合這些策略，您可以通過(guò)正念的工程和端到端的優(yōu)化來(lái)實(shí)現(xiàn)每秒數(shù)以萬(wàn)計(jì)的消息，並在適度的硬件上獲得可靠性和效率。

當(dāng)在Java中構(gòu)建可擴(kuò)展的分佈式系統(tǒng)時(shí)，消息傳遞在解耦組件和實(shí)現(xiàn)異步通信方面起著至關(guān)重要的作用。 RabbitMQ是一個(gè)廣泛採(cǎi)用的開(kāi)源消息經(jīng)紀(jì)人，在這個(gè)領(lǐng)域擅長(zhǎng)，尤其是在性能和可靠性不可傳輸?shù)那闆r下。但是，在Java中使用RabbitMQ實(shí)現(xiàn)高性能消息不僅需要基本設(shè)置。它需要智能配置，有效使用AMQP協(xié)議以及對(duì)常見(jiàn)瓶頸的認(rèn)識(shí)。

這是在高通量Java環(huán)境中充分利用RabbitMQ的方法。

1。使用連接和頻道池

RabbitMQ連接的創(chuàng)建昂貴 - 它們涉及TCP握手和AMQP協(xié)議談判。雖然Connection是線(xiàn)程安全的，但Channel對(duì)象（用於發(fā)布和消費(fèi)）不是。但是，為每個(gè)操作創(chuàng)建一個(gè)新渠道也效率也降低。

?最佳實(shí)踐：

• 在您的應(yīng)用程序上重複使用單個(gè)Connection 。
• 使用通道池（例如，與Apache Commons Pool2）來(lái)管理輕巧的可重複使用的Channel實(shí)例。
• 或者，使用諸如Spring AMQP之類(lèi)的框架，該框架可以通過(guò)CachingConnectionFactory自動(dòng)處理合併。

 CachingConnectionFactory Factory =新的CachingConnectionFactory（“ Local -Host”）;
Factory.setChannelCachesize（25）;
Factory.SetCachemode（Cachemode.Channel）;

這減少了頭頂並避免通道洩漏。

2。啟用發(fā)布者確認(rèn)並使用異步出版

默認(rèn)情況下，RabbitMQ不會(huì)確認(rèn)給經(jīng)紀(jì)人的消息傳遞。在高性能的情況下，您需要保證交付而不會(huì)犧牲速度。

?啟用發(fā)布者確認(rèn)：

• 打開(kāi)RabbitMQ中的publisher-confirms 。
• 使用channel.confirmSelect()並收聽(tīng)ConfirmListener列表回調(diào)。

但是阻止每個(gè)確認(rèn)會(huì)殺死吞吐量。

?使用異步批處理確認(rèn)：

• 發(fā)送多條消息，然後在批處理中等待確認(rèn)。
• 或使用nack感知的偵聽(tīng)器重新發(fā)布失敗的消息。

 channel.confirmselect（）;
for（int i = 0; i <messageCount; i）{
    channel.basicpublish（Exchange，RoutingKey，null，MessageBody）;
}
Channel.WaitforConfirmsordie（5_000）; //等待所有確認(rèn)

為了獲得更好的性能，軌道交付標(biāo)籤和手柄均異步確認(rèn)。

3。優(yōu)化消息大小和序列化

大消息減慢網(wǎng)絡(luò)傳輸並增加Java中的GC壓力。

?降低有效載荷尺寸：

• 避免發(fā)送冗餘數(shù)據(jù)。
• 使用有效的序列化（例如Protobuf ， Kryo或Avro ），而不是JSON或Java序列化等詳細(xì)格式。

?壓縮大消息：

• 在發(fā)送之前（例如，使用GZIP），特別是對(duì)於文本較重的數(shù)據(jù)，請(qǐng)壓縮有效載荷。
• 但是平衡CPU成本與網(wǎng)絡(luò)節(jié)省。

另外，考慮拆分大消息 - RabbitMQ具有實(shí)際限制（默認(rèn)最大消息大小?128MB，但性能降低了得多）。

4。使用消費(fèi)者預(yù)取以平衡負(fù)載

在多消費(fèi)者設(shè)置中，RabbitMQ默認(rèn)情況下使用圓形啟動(dòng)調(diào)度。沒(méi)有流量控制，快速生產(chǎn)商會(huì)壓倒消費(fèi)者的速度。

?設(shè)置預(yù)取計(jì)數(shù)：

• 使用basicQos(prefetchCount)限制每個(gè)消費(fèi)者未經(jīng)批評(píng)的消息的數(shù)量。

 Channel.basicqos（50）; //最多允許50個(gè)不添加消息

這可以使公平派遣並防止一個(gè)緩慢的消費(fèi)者阻止其他消費(fèi)者。

對(duì)於高通量消費(fèi)者，請(qǐng)根據(jù)處理速度和內(nèi)存來(lái)調(diào)整此值。

5。利用連接多路復(fù)用多個(gè)通道

您不能使用多個(gè)線(xiàn)程的單個(gè)通道，但是可以在一個(gè)連接上使用多個(gè)通道。

?使用專(zhuān)用渠道：

• 一個(gè)出版的渠道。
• 一個(gè)或多個(gè)用於消費(fèi)。
• 避免在共享渠道上阻止操作。

這有助於避免線(xiàn)路阻塞並改善並發(fā)。

6。監(jiān)視和調(diào)整兔子內(nèi)部

即使是最好的Java代碼也無(wú)法克服錯(cuò)誤配置的經(jīng)紀(jì)人。

?鑰匙兔調(diào)整提示：

• 如果使用持久性，請(qǐng)?jiān)诳焖俅鎯?chǔ)（SSD）上運(yùn)行RabbitMQ。
• 監(jiān)視隊(duì)列長(zhǎng)度 - 長(zhǎng)隊(duì)列增加了內(nèi)存使用和緩慢的恢復(fù)。
• 使用懶惰的隊(duì)列，用於較高的消息量，並且不經(jīng)常訪問(wèn)。
• 啟用鏡像隊(duì)列（或使用法定人士隊(duì)列）進(jìn)行HA，而不會(huì)犧牲太多的性能。

使用rabbitmqctl或管理插件來(lái)監(jiān)視消息率，消費(fèi)者速率和背壓。

7。優(yōu)雅處理失敗

如果您的系統(tǒng)在負(fù)載下?lián)p壞，高性能就無(wú)關(guān)緊要。

?實(shí)施：

• 重新連接邏輯具有指數(shù)向後。
• 願(yuàn)意消費(fèi)者（如果重新傳遞）。
• 下游服務(wù)的斷路器。

使用諸如Pika （Python）或經(jīng)過(guò)良好測(cè)試的Java庫(kù)（例如RabbitMQ Java客戶(hù)端或Spring AMQP）的強(qiáng)大客戶(hù)端。

最後的想法

Java中使用RabbitMQ的高性能消息並不是關(guān)於原始速度的，而是關(guān)於效率，彈性和智能資源的使用。通過(guò)組合適當(dāng)?shù)目蛻?hù)端池，異步確認(rèn)，消息優(yōu)化和經(jīng)紀(jì)人調(diào)整，即使是在適度的硬件上，您也可以每秒達(dá)到數(shù)以萬(wàn)計(jì)的消息。

這不是魔術(shù) - 只是正念的工程。

基本上，不要將兔子視為黑匣子。了解從Java應(yīng)用程序到經(jīng)紀(jì)人再到消費(fèi)者的流程，並優(yōu)化每個(gè)跳躍。

以上是與RabbitMQ的高性能Java消息傳遞的詳細(xì)內(nèi)容。更多資訊請(qǐng)關(guān)注PHP中文網(wǎng)其他相關(guān)文章！