RabbitMQ持久化機制、記憶體磁碟控制（四）
阿新 • • 發佈：2021-10-14
一、持久化
如果看到這一篇文章的朋友，都是有經驗的開發人員，對持久化的概念就不用再做過多的解析了，經過前面的幾篇文章，其實不難發現RabbitMQ 的持久化其實就只分交換器持久化、佇列持久化和訊息持久化這三個部分；
- 定義持久化交換器，通過第三個引數 durable 開啟/關閉持久化
channel.exchangeDeclare(exchangeName, exchangeType, durable)
- 定義持久化佇列，通過第二個引數 durable 開啟/關閉持久化
channel.queueDeclare(queue, durable, exclusive, autoDelete, arguments);
- 傳送持久化訊息，需要在訊息屬性中設定 deliveryMode=2 ， 此屬性在 BasicProperties 中，通過 basicPublish 方法的 props 引數傳入。
channel.basicPublish(exchange, routingKey, props, body);BasicProperties 物件可以從RabbitMQ 內建的 MessageProperties 類中獲取
MessageProperties.PERSISTENT_TEXT_PLAIN 1
如果還需要設定其它屬性，可以通過 AMQP.BasicProperties.Builder 去構建一個BasicProperties 物件；這個用法在前兩篇文章中都有展示過
newAMQP.BasicProperties.Builder() .deliveryMode(2) .build()
二、持久化程式碼演示
/** * 持久化示例 */ public class Consumer { private static Runnable receive = new Runnable() { public void run() { // 1、建立連線工廠 ConnectionFactory factory = new ConnectionFactory(); // 2、設定連線屬性factory.setHost("192.168.0.1"); factory.setUsername("admin"); factory.setPassword("admin"); Connection connection = null; Channel channel = null; final String clientName = Thread.currentThread().getName(); String queueName = "routing_test_queue"; try { // 3、從連線工廠獲取連線 connection = factory.newConnection("消費者-" + clientName); // 4、從連結中建立通道 channel = connection.createChannel(); // 定義一個持久化的，direct型別交換器 channel.exchangeDeclare("routing_test", "direct", true); // 定義一個持久化佇列 channel.queueDeclare(queueName, true, false, false, null); // 將佇列和交換器繫結，第三個引數 routingKey是關鍵，通過此路由鍵決定接收誰的訊息 channel.queueBind(queueName, "routing_test", clientName); // 定義訊息接收回調物件 DeliverCallback callback = new DeliverCallback() { public void handle(String consumerTag, Delivery message) throws IOException { System.out.println(clientName + " 收到訊息：" + new String(message.getBody(), "UTF-8")); } }; // 監聽佇列 channel.basicConsume(queueName, true, callback, new CancelCallback() { public void handle(String consumerTag) throws IOException { } }); System.out.println(clientName + " 開始接收訊息"); System.in.read(); } catch (IOException e) { e.printStackTrace(); } catch (TimeoutException e) { e.printStackTrace(); } finally { // 8、關閉通道 if (channel != null && channel.isOpen()) { try { channel.close(); } catch (IOException e) { e.printStackTrace(); } catch (TimeoutException e) { e.printStackTrace(); } } // 9、關閉連線 if (connection != null && connection.isOpen()) { try { connection.close(); } catch (IOException e) { e.printStackTrace(); } } } } }; public static void main(String[] args) { new Thread(receive, "c1").start(); new Thread(receive, "c2").start(); } }
public class Producer { public static void main(String[] args) { // 1、建立連線工廠 ConnectionFactory factory = new ConnectionFactory(); // 2、設定連線屬性 factory.setHost("192.168.0.1"); factory.setUsername("admin"); factory.setPassword("admin"); Connection connection = null; Channel channel = null; try { // 3、從連線工廠獲取連線 connection = factory.newConnection("生產者"); // 4、從連結中建立通道 channel = connection.createChannel(); // 定義一個持久化的，direct型別交換器 channel.exchangeDeclare("routing_test", "direct", true); // 記憶體、磁碟預警時用 System.out.println("按回車繼續"); System.in.read(); // 訊息內容 String message = "Hello A"; // 傳送持久化訊息到routing_test交換器上 channel.basicPublish("routing_test", "c1", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes()); System.out.println("訊息 " + message + " 已傳送！"); // 訊息內容 message = "Hello B"; // 傳送持久化訊息到routing_test交換器上 channel.basicPublish("routing_test", "c2", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes()); System.out.println("訊息 " + message + " 已傳送！"); // 記憶體、池畔預警時用 System.out.println("按回車結束"); System.in.read(); } catch (IOException e) { e.printStackTrace(); } catch (TimeoutException e) { e.printStackTrace(); } finally { // 7、關閉通道 if (channel != null && channel.isOpen()) { try { channel.close(); } catch (IOException e) { e.printStackTrace(); } catch (TimeoutException e) { e.printStackTrace(); } } // 8、關閉連線 if (connection != null && connection.isOpen()) { try { connection.close(); } catch (IOException e) { e.printStackTrace(); } } } } }
二、記憶體告警
預設情況下 set_vm_memory_high_watermark 的值為 0.4，即記憶體閾值（臨界值）為 0.4，表示當RabbitMQ 使用的記憶體超過 40%時，就會產生記憶體告警並阻塞所有生產者的連線。一旦告警被解除(有訊息被消費或者從記憶體轉儲到磁碟等情況的發生)， 一切都會恢復正常。在出現記憶體告警後，所有的客戶端連線都會被阻塞。阻塞分為 blocking 和 blocked 兩種。- blocking：表示沒有傳送訊息的連結。
- blocked：表示試圖傳送訊息的連結。
rabbitmqctl set_vm_memory_high_watermark 1 1或者
rabbitmqctl set_vm_memory_high_watermark absolute 1GB但這種方式只是臨時調整，RabbitMQ 服務重啟後，會還原。如果需要永久調整，可以修改配置檔案。但修改配置檔案需要重啟RabbitMQ 服務才能生效。 修改配置檔案： vim /etc/rabbitmq/rabbitmq.conf
vm_memory_high_watermark.relative = 0.4 1或者
vm_memory_high_watermark.absolute = 1GB
三、模擬記憶體告警
1. 調整記憶體閾值，模擬出告警，在RabbitMQ 伺服器上修改。 注意：修改之前，先在管理頁面看一下當前使用了多少，調成比當前值小rabbitmqctl set_vm_memory_high_watermark absolute 50MB
2.重新整理管理頁面（可能需要重新整理多次），在 Overview -> Nodes 中可以看到Memory變成了紅色，表示此節點記憶體告警了
3. 啟動 Producer 和 Consumer（原始碼連結在最下面） 4. 檢視管理介面的 Connections 頁面，可以看到生產者和消費者的連結都處於 blocking 狀態。 5. 在 Producer 的控制檯按回車健，再觀察管理介面的 Connections 頁面，會發現生產者的狀態成了 blocked 。 6. 此時雖然在 Producer 控制檯看到了傳送兩條訊息的資訊，但 Consumer 並沒有收到任何訊息。並且在管理介面的 Queues 頁面也看到不到佇列的訊息數量有變化。 7. 解除記憶體告警後，會發現 Consumer 收到了 Producer 傳送的兩條訊息。四、記憶體換頁
- 在Broker節點的使用記憶體即將達到記憶體閾值之前，它會嘗試將佇列中的訊息儲存到磁碟以釋放記憶體空間，這個動作叫記憶體換頁。
- 持久化和非持久化的訊息都會被轉儲到磁碟中，其中持久化的訊息本身就在磁碟中有一份副本，此時會將持久化的訊息從記憶體中清除掉。
- 預設情況下，在記憶體到達記憶體閾值的 50%時會進行換頁動作。也就是說，在預設的記憶體閾值為 0.4的情況下，當記憶體超過 0.4 x 0 .5=0.2 時會進行換頁動作。
- 通過修改配置檔案，調整記憶體換頁分頁閾值（不能通過命令調整）。
# 此值大於1時，相當於禁用了換頁功能。 vm_memory_high_watermark_paging_ratio = 0.75
五、磁碟告警
- 當磁碟剩餘空間低於磁碟的閾值時，RabbitMQ 同樣會阻塞生產者，這樣可以避免因非持久化的訊息持續換頁而耗盡磁碟空間導致服務崩潰
- 預設情況下，磁碟閾值為50MB，表示當磁碟剩餘空間低於50MB 時會阻塞生產者並停止記憶體中訊息的換頁動作
- 這個閾值的設定可以減小，但不能完全消除因磁碟耗盡而導致崩漬的可能性。比如在兩次磁碟空間檢測期間內，磁碟空間從大於50MB被耗盡到0MB
- 通過命令可以調整磁碟閾值，臨時生效，重啟恢復
# disk_limit 為固定大小，單位為MB、GB
rabbitmqctl set_disk_free_limit <disk_limit>
或者
# fraction 為相對比值，建議的取值為1.0~2.0之間 rabbitmqctl set_disk_free_limit mem_relative <fraction>
其實這些內容在官網上都有說明，有興趣可以直接看官網：https://www.rabbitmq.com/alarms.html
這短短的一生我們最終都會失去，不妨大膽一點，愛一個人，攀一座山，追一個夢