RabbitMQ的訊息佇列是駐留記憶體的, 萬一機器Down機的話, 記憶體裡的佇列, 也會消失不見. 為解決這個問題, 才有了”持久化” 這個概念.
顧名思義: 將訊息佇列存到硬碟上(落盤)的檔案中, 勢必會影響效率

訊息實體

package com.yuchen.demo.falldisc;

import java.io.Serializable;

public class Message implements Serializable {

    private static final long serialVersionUID = 1L;
    String xxx;

    public
 
 Message(String xxx) {
        this.xxx = xxx;
    }

    public String getXxx() {
        return xxx;
    }

    public void setXxx(String xxx) {
        this.xxx = xxx;
    }

    @Override
    public String toString() {
        return this.xxx;
    }


}

生產者

package com.yuchen.demo.falldisc;
 


import java.io.ByteArrayOutputStream;
import java.io.ObjectOutputStream;

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

public class Send {

    // 佇列名稱
    private final static String QUEUE_NAME = "lsy_falldisc"
 
;


    public static void main(String[] args) throws Exception {

        // 建立連線連線到MabbitMQ
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("10.13.144.24");
        factory.setVirtualHost("/");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("admin");
        // 新建連線
        Connection connection = factory.newConnection();
        // 建立通道
        Channel channel = connection.createChannel();

        // 建立佇列 (佇列名稱, 持久化, 僅有1個consumer 與autoDelete配合使用 沒有consumer自動刪除 , 自動刪除, 其他引數)
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);

        // 傳送訊息 byte[]
        // 使用者可以把自己的訊息序列化成JSON等格式在轉成byte[]傳送到佇列中取出訊息後再反序列化得到訊息內容
        Message message = new Message("Hello Message !");
        // 訊息持久化 MessageProperties.PERSISTENT_TEXT_PLAIN
        ByteArrayOutputStream baos = new ByteArrayOutputStream(10000);
        ObjectOutputStream oos = new ObjectOutputStream(baos);
        oos.writeObject(message);
        channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, baos.toByteArray());
        System.out.println("傳送成功: '" + message + "'");

        // 關閉連線
        channel.close();
        connection.close();


    }
}

消費者

package com.yuchen.demo.falldisc;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

public class MessageConsumer {

    // 佇列名稱
    private final static String QUEUE_NAME = "message_falldisc";

    public static void start() throws Exception {

        // 建立連線連線到MabbitMQ
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("10.13.144.24");
        factory.setVirtualHost("/");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("admin");
        // 新建連線
        Connection connection = factory.newConnection();
        // 建立通道
        Channel channel = connection.createChannel();

        // 宣告佇列, 可能在傳送方之前啟動接收方, 確保佇列存在。
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                ByteArrayInputStream bais = new ByteArrayInputStream(body);
                ObjectInputStream ois = new ObjectInputStream(bais);
                Message message = null;
                try {
                    message = (Message) ois.readObject();
                } catch (ClassNotFoundException e) {
                    e.printStackTrace();
                }
                System.out.println("接收到遠端訊息: '" + message.toString() + "'");
            }
        };
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }

}

重點在於
佇列要持久化(第二個引數 true)

    channel.queueDeclare(QUEUE_NAME, true, false, false, null);

訊息也要持久化(第三個引數MessageProperties.PERSISTENT_TEXT_PLAIN)

channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, baos.toByteArray());