rabbitmq學習記錄

作用

訊息佇列，實現解耦。實現業務系統間的通訊。

搭建過程

在之前搭好的伺服器上按照ubuntu18.04安裝rabbitmq的方式安裝即可。安裝成功後再ip:埠可以直接訪問圖形介面。java這邊在maven裡把依賴加上就ok了。

學習過程

maven依賴

    <dependencies>
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.8.0</version>
        </dependency>
        <dependency>
            <groupId>commons-io</groupId>
            <artifactId>commons-io</artifactId>
            <version>2.6</version>
        </dependency>
    </dependencies> 
 

簡單程式實現資訊傳遞

生產者

public class Producer {
    //佇列名稱
    public static final String QUEUE_NAME = "hello";
    public static void main(String[] args) throws Exception{
        //建立連線工廠
        ConnectionFactory factory = new ConnectionFactory();
        //工廠ip
        factory.setHost("202.200.231.14");
        //使用者名稱
        factory.setUsername("admin");
        //密碼
        factory.setPassword("admin");

        //建立連線
        Connection connection = factory.newConnection();
        //獲取通道
        Channel channel = connection.createChannel();
        /**
         * 1.佇列名稱
         * 2.訊息是否持久化
         * 3.是否能有多個消費者
         * 4.是否自動刪除
         */
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        String message = "hello world";

        channel.basicPublish("",QUEUE_NAME,null,message.getBytes(StandardCharsets.UTF_8));
        System.out.println("訊息傳送成功");
    }
} 
 

消費者

public class Consumer {
    //佇列的名稱
    public static final String QUEUE_NAME = "hello";
    public static void main(String[] args) throws  Exception{
        //建立工廠
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("202.200.231.14");
        factory.setUsername("admin");
        factory.setPassword("admin");
        Connection connection = factory.newConnection();

        Channel channel = connection.createChannel();

        DeliverCallback deliverCallback = (consumerTag,message) ->{
            System.out.println(new String(message.getBody()));
        };

        CancelCallback cancelCallback = (consumerTag) ->{
            System.out.println("訊息消費被中斷");
        };
        /**
         * 1.消費哪個佇列
         * 2.消費成功後是否需要自動應答
         * 3.消費者未成功消費的回撥
         * 4.消費者取消消費的的回撥
         */

        channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
    }
}
 

工作佇列模式

輪詢，多個消費者執行緒，處理資訊。直接多開消費者，可以輪流接收訊息。

抽取工具類 rabbitimqutils

public class RabbitMqUtils {
    public static Channel getChannel() throws Exception{
        //建立連線工廠
        ConnectionFactory factory = new ConnectionFactory();
        //工廠ip
        factory.setHost("202.200.231.14");
        //使用者名稱
        factory.setUsername("admin");
        //密碼
        factory.setPassword("admin");

        //建立連線
        Connection connection = factory.newConnection();
        //獲取通道
        Channel channel = connection.createChannel();
        return channel;
    }
}

訊息應答手動版

/**
 * 訊息再手動應答時不丟失，重新放回到佇列
 */
public class Worker3 {
    public static final String task_queue_name ="ack_queue";
    public static void main(String[] args) throws Exception{
        Channel channel = RabbitMqUtils.getChannel();
        System.out.println("C1很快");

        DeliverCallback deliverCallback = (consumerTag, message) ->{
            //沉睡1s
            SleepUtils.sleep(1);
            System.out.println("接收到的訊息" + new String(message.getBody()));
            //手動應答
            /**
             * 1. 訊息的標記tag
             * 2. 是否批量應答訊息
             */
            channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
        };

        CancelCallback cancelCallback = (consumerTag) ->{
            System.out.println(""+"拔劍四顧心茫然");
        };

        channel.basicConsume(task_queue_name,false,deliverCallback,cancelCallback);

    }
}