RabbitMQ使用SpringAMQP的配置方法

更新時間：2024年03月07日 10:33:30 作者：在無清風(fēng)

這篇文章主要介紹了RabbitMQ使用SpringAMQP的配置方法,本文給大家介紹的非常詳細,對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值,需要的朋友參考下吧

簡介

絕對的簡單，絕對的易懂，方便初學(xué)者，更加利于理解和上手使用（代碼可直接復(fù)制粘貼進行使用）

如有其它問題，大家可以留言或私聊。

主要為了給大家展示各個代碼使用

如果需要更加完整的文檔，可以點擊下方連接進行閱讀http://t.csdnimg.cn/WWdhG

配置

消息發(fā)送配置

logging:
  pattern:
    dateformat: MM-dd HH:mm:ss:SSS
spring:
  rabbitmq:
    host: 8.137.59.245
    port: 5672
    username: itcast
    password: 123321
    # 設(shè)置虛擬主機路徑
    virtual-host: /

消息接收配置

logging:
  pattern:
    dateformat: MM-dd HH:mm:ss:SSS
spring:
  rabbitmq:
    host: 8.137.59.245
    port: 5672
    username: itcast
    password: 123321
    virtual-host: /
    listener:
      simple:
        prefetch: 1 #每次只能接收一個消息，處理完后在隊列中刪除，在獲取下一個消息

Basic Queue簡單隊列

簡單介紹

消息發(fā)送者直接發(fā)送消息給隊列，消息接收者接收隊列中發(fā)過來的信息，實現(xiàn)消息的接收。

消息發(fā)送者

//    Basic Queue簡單隊列
    @Test
    public void testSendMessage2SimpleQueue(){
        //隊列名稱
        String queueName = "simple.queue";
        //消息
        String message = "Hello World!spring amqp!!";
        //發(fā)送消息
        rabbitTemplate.convertAndSend(queueName, message);
        /**
         *
         *rabbitTemplate.convertAndSend(queueName, routingKey, message);
         * 發(fā)送消息，參數(shù)分別是：隊列名稱，RoutingKey(暫時為空)，消息
         */
    }

消息接收者

// Basic Queue簡單隊列
    @RabbitListener(queues = "simple.queue")
    public void listenSimpleQueue(String msg){
        System.out.println("消費者接收到simple.queue的消息：{"+msg+"}");
//        通過simple.queue隊列接收消息，并答應(yīng)出接收的消息
    }

Work Queue隊列

簡單介紹

Work Queue隊列和Basic Queue隊列比較。當發(fā)送多個信息需要處理Work Queue隊列可以分別交給多個隊列處理，增加處理消息速度

消息發(fā)送者

    //WorkQueue隊列
    @Test
    public void testSendMessage2WorkQueue() throws InterruptedException {
        // 隊列名稱
        String queueName = "simple.queue";
        // 消息
        String message = "Hello World!spring amqp!!__";
        for (int i = 1; i < 50; i++) {
            // 發(fā)送消息
            rabbitTemplate.convertAndSend(queueName, message+i);
            Thread.sleep(20);
        }
//        發(fā)送50個消息到simple.queue隊列中
    }

消息接收者

 
// Work Queue工作隊列
    @RabbitListener(queues = "simple.queue")
    public void listenWorkQueue1(String msg) throws InterruptedException {
        System.out.println("消費者1接收到simple.queue的消息：{"+msg+"}"+ LocalTime.now());
        Thread.sleep(20);
    }
    @RabbitListener(queues = "simple.queue")
    public void listenWorkQueue2(String msg) throws InterruptedException {
        System.out.println("消費者2接收到simple.queue的消息：{"+msg+"}"+ LocalTime.now());
        Thread.sleep(50);
    }
//    分別用兩個消息接收者來接收消息，防止同一個隊列處理太快，所以使用sleep來減慢處理速度

發(fā)布訂閱模型-Fanout交換機

簡單介紹

這里添加交換機（主要將不同消息推送到不同隊列中進行消費）

消息發(fā)送者

    // 發(fā)布訂閱模型-Fanout交換機
    @Test
    public void testSendFanoutExchange(){
        //交換機名稱
        String exchangeName = "itcast.fanout";
        //消息
        String message = "hello,every one!";
        //發(fā)送消息，參數(shù)分別是：交換機名稱，RoutingKey(暫時為空)，消息
        rabbitTemplate.convertAndSend(exchangeName, "", message);
//        這里我們不給隊列名，而給交換機名稱，通過交換機來發(fā)送消息給不同的隊列
    }

消息接收者

這里添加了交換機所以我們先寫交換機和不同的隊列進行綁定

這里我們使用類來操作，后面兩不會使用這種比較繁瑣的綁定方式

交換機

    //itcast.fanout(交換機)
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("itcast.fanout");
    }
    //itcast.queue1（隊列一）
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");
    }
    //綁定隊列1到交換機
    @Bean
    public Binding fanoutBinding1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
        return BindingBuilder
                .bind(fanoutQueue1).
                to(fanoutExchange);
    }
    //itcast.queue2（隊列二）
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }
    //綁定隊列2到交換機
    @Bean
    public Binding fanoutBinding2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
        return BindingBuilder
               .bind(fanoutQueue2).
                to(fanoutExchange);
    }
/**
 *  這里先定義一個隊列，用于接收消息發(fā)送者發(fā)送的消息
 *  然后再定義一個交換機，用于接收隊列1和隊列2的消息
 *  這里的交換機和前面消息發(fā)送者的交換機一樣名字。
 */

消息接收者

// 發(fā)布訂閱模型-Fanout交換機
    @RabbitListener(queues = "fanout.queue1")
    public void listenFanoutQueue1(String msg){
        System.out.println("消費者1接收到fanout.queue1的消息：{"+msg+"}");
    }
    @RabbitListener(queues = "fanout.queue2")
    public void listenFanoutQueue2(String msg){
        System.out.println("消費者2接收到fanout.queue2的消息：{"+msg+"}");
    }

訂閱發(fā)布模型-Direct交換機

簡單介紹

Direct交換機和Fanout交換機比較，不一樣點在于RoutingKey給的值不一樣

消息發(fā)送者

    // 訂閱，發(fā)布模型-Direct交換機
    @Test
    public void testSendDirectExchange(){
        //交換機名稱
        String exchangeName = "itcast.direct";
        //消息
        String message = "hello,blue one!";
        //發(fā)送消息，參數(shù)分別是：交換機名稱，RoutingKey(暫時為空)，消息
        rabbitTemplate.convertAndSend(exchangeName, "blue", message);
        /**
         * 消息發(fā)送，參數(shù)分別是：交換機名稱，RoutingKey，消息
         * routingKey: 用來區(qū)分不同的隊列
         * 上面routingKey給“blue”的意義是將消息發(fā)送給blue隊列
         * */
    }

消息接收者

// 訂閱，發(fā)布模型-Direct交換機
    @RabbitListener(bindings = @QueueBinding(
            //隊列
            value = @Queue(name = "direct.queue1"),
            //交換機
            exchange = @Exchange(name = "itcast.direct",type = ExchangeTypes.DIRECT),
            //邦定機置
            key = {"red","blue"}
    ))
    public void listenDirectQueue1(String msg){
        System.out.println("消費者接收到direct.queue1的消息：{"+msg+"}");
//        接收red和blue的消息并傳入隊列進行消費
    }
    @RabbitListener(bindings = @QueueBinding(
            //隊列
            value = @Queue(name = "direct.queue2"),
            //交換機
            exchange = @Exchange(name = "itcast.direct",type = ExchangeTypes.DIRECT),
            //邦定機置
            key = {"red","yellow"}
    ))
    public void listenDirectQueue2(String msg){
        System.out.println("消費者接收到direct.queue2的消息：{"+msg+"}");
//        接收red和yellow的消息并傳入隊列進行消費
    }

發(fā)布訂閱模型-Topic交換機

簡單介紹

個人理解：算是對Direct交換機進行擴展吧，可以實現(xiàn)更多樣性的訂閱發(fā)布

消息發(fā)送者

    // 訂閱，發(fā)布模型-Direct交換機
    @Test
    public void testSendDirectExchange(){
        //交換機名稱
        String exchangeName = "itcast.direct";
        //消息
        String message = "hello,blue one!";
        //發(fā)送消息，參數(shù)分別是：交換機名稱，RoutingKey(暫時為空)，消息
        rabbitTemplate.convertAndSend(exchangeName, "blue", message);
        /**
         * 消息發(fā)送，參數(shù)分別是：交換機名稱，RoutingKey，消息
         * routingKey: 用來區(qū)分不同的隊列
         * 上面routingKey給“blue”的意義是將消息發(fā)送給blue隊列
         * */
    }

消息接收者

// 訂閱，發(fā)布模型-Direct交換機
    @RabbitListener(bindings = @QueueBinding(
            //隊列
            value = @Queue(name = "direct.queue1"),
            //交換機
            exchange = @Exchange(name = "itcast.direct",type = ExchangeTypes.DIRECT),
            //邦定機置
            key = {"red","blue"}
    ))
    public void listenDirectQueue1(String msg){
        System.out.println("消費者接收到direct.queue1的消息：{"+msg+"}");
//        接收red和blue的消息并傳入隊列進行消費
    }
    @RabbitListener(bindings = @QueueBinding(
            //隊列
            value = @Queue(name = "direct.queue2"),
            //交換機
            exchange = @Exchange(name = "itcast.direct",type = ExchangeTypes.DIRECT),
            //邦定機置
            key = {"red","yellow"}
    ))
    public void listenDirectQueue2(String msg){
        System.out.println("消費者接收到direct.queue2的消息：{"+msg+"}");
//        接收red和yellow的消息并傳入隊列進行消費
    }

Json序列化

簡單介紹

為什么要添加呢：大致就是實現(xiàn)可消息發(fā)送類型的更多

添加Maven

        <!--rabbitmq使用json序列化-->
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
        </dependency>

消息發(fā)送者

    // 使用json發(fā)送對象隊列
    @Test
    public void testSendObjectQueue() {
        Map<String, Object> msg = new HashMap<>();
        msg.put("name", "留言");
        msg.put("age", 21);
        rabbitTemplate.convertAndSend("object.queue", msg);
//        發(fā)送Map類型的消息到object.queue隊列中
    }

消息接收者

    @RabbitListener(queues = "object.queue")
    public void listenObjectQueue(Map<String,Object> msg){
        System.out.println("消費者接收到object.queue的消息：{"+msg+"}");
        msg.forEach((k,v)->{
            System.out.println(k+"="+v);
            System.out.println("{"+k+"}");
        });
//        接收消息并循環(huán)打印接收到的消息
    }