一、Virtual Hosts

public class Producer {
    public static void main(String[] args) {
        // 1、創建連接工廠
        ConnectionFactory factory = new ConnectionFactory();
        // 2、設置連接屬性
        factory.setUsername("order-user");
        factory.setPassword("order-user");
        factory.setVirtualHost("v1");

        Connection connection = null;
        Channel channel = null;

        // 3、設置每個節點的鏈接地址和埠
        Address[] addresses = new Address[]{
                new Address("192.168.0.1", 5672),
                new Address("192.168.0.2", 5672)
        };

        try {
            // 開啟/關閉連接自動恢復，默認是開啟狀態
            factory.setAutomaticRecoveryEnabled(true);

            // 設置每100毫秒嘗試恢復一次，默認是5秒：com.rabbitmq.client.ConnectionFactory.DEFAULT_NETWORK_RECOVERY_INTERVAL
            factory.setNetworkRecoveryInterval(100);

            factory.setTopologyRecoveryEnabled(false);

            // 4、使用連接集合裡面的地址獲取連接
            connection = factory.newConnection(addresses, "生產者");

            // 添加重連監聽器
            ((Recoverable) connection).addRecoveryListener(new RecoveryListener() {
                /**
                 * 重連成功後的回調
                 * @param recoverable
                 */
                public void handleRecovery(Recoverable recoverable) {
                    System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SS").format(new Date()) + " 已重新建立連接！");
                }

                /**
                 * 開始重連時的回調
                 * @param recoverable
                 */
                public void handleRecoveryStarted(Recoverable recoverable) {
                    System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SS").format(new Date()) + " 開始嘗試重連！");
                }
            });

            // 5、從鏈接中創建通道
            channel = connection.createChannel();

            /**
             * 6、聲明（創建）隊列
             * 如果隊列不存在，才會創建
             * RabbitMQ 不允許聲明兩個隊列名相同，屬性不同的隊列，否則會報錯
             *
             * queueDeclare參數說明：
             * @param queue 隊列名稱
             * @param durable 隊列是否持久化
             * @param exclusive 是否排他，即是否為私有的，如果為true,會對當前隊列加鎖，其它通道不能訪問，並且在連接關閉時會自動刪除，不受持久化和自動刪除的屬性控制
             * @param autoDelete 是否自動刪除，當最後一個消費者斷開連接之後是否自動刪除
             * @param arguments 隊列參數，設置隊列的有效期、消息最大長度、隊列中所有消息的生命周期等等
             */
            channel.exchangeDeclare("test-exchange", "fanout");
            channel.queueDeclare("queue1", false, false, false, null);
            channel.queueBind("queue1", "test-exchange", "xxoo");
            for (int i = 0; i < 100; i++) {
                // 消息內容
                String message = "Hello World " + i;
                try {
                    // 7、發送消息
                    channel.basicPublish("test-exchange", "queue1", null, message.getBytes());
                } catch (AlreadyClosedException e) {
                    // 可能連接已關閉，等待重連
                    System.out.println("消息 " + message + " 發送失敗！");
                    i--;
                    TimeUnit.SECONDS.sleep(2);
                    continue;
                }
                System.out.println("消息 " + i + " 已發送！");
                TimeUnit.SECONDS.sleep(2);
            }

        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        } catch (InterruptedException 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 class VirtualHosts {
    public static void main(String[] args) {
        // 1、創建連接工廠
        ConnectionFactory factory = new ConnectionFactory();
        // 2、設置連接屬性
        factory.setUsername("order-user");
        factory.setPassword("order-user");
        factory.setVirtualHost("v1");

        Connection connection = null;
        Channel prducerChannel = null;
        Channel consumerChannel = null;

        // 3、設置每個節點的鏈接地址和埠
        Address[] addresses = new Address[]{
                new Address("192.168.0.1", 5672),
                new Address("192.168.0.2", 5672)
        };

        try {
            // 4、從連接工廠獲取連接
            connection = factory.newConnection(addresses, "消費者");

            // 5、從鏈接中創建通道
            prducerChannel = connection.createChannel();

            prducerChannel.exchangeDeclare("test-exchange", "fanout");
            prducerChannel.queueDeclare("queue1", false, false, true, null);
            prducerChannel.queueBind("queue1", "test-exchange", "xxoo");
            // 消息內容
            String message = "Hello A";
            prducerChannel.basicPublish("test-exchange", "c1", null, message.getBytes());

            consumerChannel = connection.createChannel();
            // 創建一個消費者對象
            Consumer consumer = new DefaultConsumer(consumerChannel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("收到消息：" + new String(body, "UTF-8"));
                }
            };
            consumerChannel.basicConsume("queue1", true, consumer);

            System.out.println("等待接收消息");
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        } finally {
            // 9、關閉通道
            if (prducerChannel != null && prducerChannel.isOpen()) {
                try {
                    prducerChannel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            }

            // 10、關閉連接
            if (connection != null && connection.isOpen()) {
                try {
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}