如何進行RabbitMq的簡單使用

這期內容當中小編將會給大家帶來有關如何進行RabbitMq的簡單使用，文章內容豐富且以專業的角度為大家分析和敘述，閱讀完這篇文章希望大家可以有所收獲。

1.pom文件中加入依賴

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId><version>2.3.3.RELEASE</version></dependency>

2.配置文件，配置mq

自動配置信息  這里我開啟ACK消息確認server.port=8088#服務器配置spring.application.name=rabbitmq-test-sending#rabbitmq連接參數spring.rabbitmq.host=localhostspring.rabbitmq.port=5672spring.rabbitmq.username=guest
spring.rabbitmq.password=guest# 開啟發送確認spring.rabbitmq.publisher-confirms=true# 開啟發送失敗退回spring.rabbitmq.publisher-returns=true# 開啟ACKspring.rabbitmq.listener.direct.acknowledge-mode=manual

3.Rabbit配置類，使用topic交換器，使用通配符，一個交換器對應多個queue

import org.springframework.amqp.core.*;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import java.util.HashMap;import java.util.Map;@Configurationpublic class RabbitmqConfig {//隊列    @Bean    public Queue queueTest1(){return new Queue("queueTest1",true);
    }/*    * 設置消息隊列的TTL（過期時間）    * */    @Bean    public Queue queueTest2(){/**         * 隊列的名稱，是否持久化，是否獨享、排外的，是否自動刪除,隊列的其他屬性參數         * （1）x-message-ttl：消息的過期時間，單位：毫秒；         * （2）x-expires：隊列過期時間，隊列在多長時間未被訪問將被刪除，單位：毫秒；         * （3）x-max-length：隊列最大長度，超過該最大值，則將從隊列頭部開始刪除消息；         * （4）x-max-length-bytes：隊列消息內容占用最大空間，受限于內存大小，超過該閾值則從隊列頭部開始刪除消息；         * （5）x-overflow：設置隊列溢出行為。這決定了當達到隊列的最大長度時消息會發生什么。有效值是drop-head、reject-publish或reject-publish-dlx。         * （6）x-dead-letter-exchange：死信交換器名稱，過期或被刪除（因隊列長度超長或因空間超出閾值）的消息可指定發送到該交換器中；         * （7）x-dead-letter-routing-key：死信消息路由鍵，在消息發送到死信交換器時會使用該路由鍵，如果不設置，則使用消息的原來的路由鍵值         * （8）x-single-active-consumer：表示隊列是否是單一活動消費者，true時，注冊的消費組內只有一個消費者消費消息，其他被忽略，false時消息循環分發給所有消費者(默認false)         * （9）x-max-priority：隊列要支持的最大優先級數;如果未設置，隊列將不支持消息優先級；         * （10）x-queue-mode（Lazy mode）：將隊列設置為延遲模式，在磁盤上保留盡可能多的消息，以減少RAM的使用;如果未設置，隊列將保留內存緩存以盡可能快地傳遞消息；         * （11）x-queue-master-locator：在集群模式下設置鏡像隊列的主節點信息。         */        Map<String, Object> arguments = new HashMap<>();
        arguments.put("x-message-ttl", 5000);return new Queue("queueTest2", true, false, false, arguments);
    }//交換機    @Bean    public TopicExchange exchangeTest(){//可以傳exchange名字，是否支持持久化，是否可以自動刪除        return new TopicExchange("exchangeTest",true,false);
    }@Bean    public Binding bindQueueTest1AndExchange(){return BindingBuilder.bind(queueTest1()).to(exchangeTest()).with("phone.routing.*");

    }@Bean    public Binding bindQueueTest2AndExchange(){return BindingBuilder.bind(queueTest2()).to(exchangeTest()).with("web.routing.*");

    }
}

4.生產者

import org.springframework.amqp.core.Message;import org.springframework.amqp.rabbit.connection.CorrelationData;import org.springframework.amqp.rabbit.core.RabbitTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Service;import java.util.Date;import java.util.UUID;/** 生產者，帶消息確認* */@Servicepublic class PruSender implements RabbitTemplate.ReturnCallback {@Autowired    private RabbitTemplate rabbitTemplate;//routing_key,把消息發送到相應的隊列中    public void sendMessage(String routing_key){//發送內容        String context = "你好現在是 " + new Date();this.rabbitTemplate.setReturnCallback(this);//發送失敗退回        this.rabbitTemplate.setConfirmCallback((correlationData,ack,message)->{//手動發送消息確認            if(!ack){
                System.out.println("消息發送失敗" + message + correlationData.toString());
            }else{
                System.out.println("消息發送成功" + correlationData.toString());
            }
        });
        CorrelationData correlationData = new CorrelationData();
        correlationData.setId(UUID.randomUUID().toString());//交換機名稱、routingKey、內容、消息Id        this.rabbitTemplate.convertAndSend("exchangeTest",routing_key, context,correlationData);
    }@Override    public void returnedMessage(Message message, int i, String s, String s1, String s2) {

        System.out.println("sender return success" + message.toString() + "===" + i + "===" + s1 + "===" + s2);
    }
}

5.消費者

import com.rabbitmq.client.Channel;import org.springframework.amqp.core.Message;import org.springframework.amqp.rabbit.annotation.RabbitHandler;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Service;import java.io.IOException;/** 消費者，帶消息確認** */@Service@RabbitListener(queues = "queueTest")public class Receiver {//消息內容，通道，消息的屬性信息    @RabbitHandler    public void immediateProcess(String text,Channel channel,Message message) throws IOException {try {
            System.out.println("Receiver" + text);/**             * 手動確認，通知mq已經成功消費改條信息，可以刪除了             * //消息的標識，false只確認當前一個消息收到，true確認所有consumer獲得的消息             */            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);

        } catch (IOException e) {
            e.printStackTrace();/*            *消費消息失敗            * 第二個參數是否應用于多消息，第三個參數是否從新計入隊列            * */            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,true);
        }


    }

}

交換機類型：
Fanout：廣播，將消息交給所有綁定到交換機的隊列
Direct：定向，把消息交給符合指定routing key 的隊列
Topic：通配符，把消息交給符合routing pattern（路由模式） 的隊列

上述就是小編為大家分享的如何進行RabbitMq的簡單使用了，如果剛好有類似的疑惑，不妨參照上述分析進行理解。如果想知道更多相關知識，歡迎關注億速云行業資訊頻道。