快捷導(dǎo)航

SpringBoot實(shí)現(xiàn)RabbitMQ三種使用方式

更新時(shí)間：2023年07月09日 12:12:44 作者：流殤?

本文主要介紹了SpringBoot實(shí)現(xiàn)RabbitMQ三種使用方式，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧

基于API的方式

1.使用AmqpAdmin定制消息發(fā)送組件

     @Autowired
    private AmqpAdmin amqpAdmin;
    @Test
    public void amqpAdmin(){
        //1.定義fanout類型的交換器
        amqpAdmin.declareExchange(new FanoutExchange("fanout_exchange"));
        //2.定義兩個(gè)默認(rèn)持久化隊(duì)列,分別處理email和sms
        amqpAdmin.declareQueue(new Queue("fanout_queue_email"));
        amqpAdmin.declareQueue(new Queue("fanout_queue_sms"));
        //3.將隊(duì)列分別與交換器進(jìn)行綁定
                                                //               隊(duì)列名             是隊(duì)列                                 交換機(jī)的名稱           路由         其它參數(shù)
        amqpAdmin.declareBinding(new Binding("fanout_queue_email",Binding.DestinationType.QUEUE,"fanout_exchange","",null));
        amqpAdmin.declareBinding(new Binding("fanout_queue_sms",Binding.DestinationType.QUEUE,"fanout_exchange","",null));
    }

2.消息發(fā)送者發(fā)送消息

創(chuàng)建實(shí)體類

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@AllArgsConstructor
@NoArgsConstructor
public class User {
    private Integer id;
    private String name;
}

發(fā)送消息

    @Autowired
    private RabbitTemplate re;
    @Test//消息發(fā)送者
    public void subPublisher(){
        User user = new User(1,"小滿");
        re.convertAndSend("fanout_exchange", "", user);
    }

如圖所以,如果我們直接發(fā)送的話就會(huì)報(bào)這個(gè)錯(cuò),有兩種解決方法,第一種是比較常用的讓實(shí)體類User實(shí)現(xiàn)序列化Serializable接口,這里我們不做演示,第二種是寫一個(gè)配置類,只有在RabbitMQ可以使用

import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitMQConfig {
    //定制JSON格式的消息轉(zhuǎn)化器
    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
}

加上配置類后我們發(fā)送就不會(huì)報(bào)錯(cuò)了,我們也可以在RabbitMQ的可視化端口看到我們發(fā)送的消息

3.發(fā)送完消息后接下來(lái)就是消費(fèi)消息了,定義接收消息的業(yè)務(wù)

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;
@Service
public class RabbitMQService {
    //發(fā)布訂閱模式: @RabbitListener可以指定當(dāng)前方法監(jiān)控哪一個(gè)隊(duì)列
   @RabbitListener(queues = "fanout_queue_email")//消費(fèi)者可以消費(fèi)多個(gè)隊(duì)列的消息 
    public void subConsumerEmail(Message message){
        //當(dāng)隊(duì)列中有內(nèi)容是方法會(huì)自動(dòng)執(zhí)行   推薦Object來(lái)接收
        //官網(wǎng)推薦Message
        byte[] body = message.getBody();//Message將數(shù)據(jù)存放在body中
        String msg = new String(body);
        System.out.println("郵件業(yè)務(wù)接收到消息:"+msg);
    }
    @RabbitListener(queues = "fanout_queue_sms") 
    public void subConsumerSms(Message message){
        byte[] body = message.getBody();
        String msg = new String(body);
        System.out.println("短信業(yè)務(wù)接收到消息:"+msg);
    }
}

4.重新運(yùn)行發(fā)送端就可以接收到我們發(fā)送的數(shù)據(jù),接收的數(shù)據(jù)可能打印在任意一個(gè)控制臺(tái)中,這是idea的機(jī)制,我們不需要管

基于配置類的方式

1.在config配置類中定義

 
import org.springframework.amqp.core.*;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitMQConfig {
    //定制JSON格式的消息轉(zhuǎn)化器
    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
   // 1.fanout創(chuàng)建一個(gè)交換機(jī)
    @Bean
    public Exchange fanoutExchange(){
        return ExchangeBuilder.fanoutExchange("fanout_exchange").build();
    }
    //2.定義消息隊(duì)列
    @Bean
    public Queue fanoutQueueEmail(){
        return new Queue("fanout_queue_email");
    }
    @Bean
    public Queue fanoutQueueSms(){
        return new Queue("fanout_queue_sms");
    }
    //3.將創(chuàng)建的隊(duì)列綁定到對(duì)應(yīng)的交換機(jī)上
    @Bean
    public Binding bingingEmail(){
        return BindingBuilder.bind(fanoutQueueEmail()).to(fanoutExchange()).with("").noargs();
    }
    @Bean
    public Binding bingingSms(){
        return BindingBuilder.bind(fanoutQueueSms()).to(fanoutExchange()).with("").noargs();
    }
}

2.為了避免api的影響,我們可以在可視化端口將基于api創(chuàng)建的交換機(jī)和隊(duì)列刪除

1)刪除交換機(jī)

2)刪除隊(duì)列,前面也是點(diǎn)擊隊(duì)列的名字

可以看到我已經(jīng)將交換機(jī)和消息隊(duì)列都已經(jīng)刪除,接下來(lái)我們重新啟動(dòng)項(xiàng)目 ,配置類可以在啟動(dòng)的時(shí)候自動(dòng)創(chuàng)建

我們的訂閱發(fā)布模式也是可以正常運(yùn)行

基于注解類的方式

1.我們要現(xiàn)將基于配置類的方式注釋掉,避免影響我們測(cè)試

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;
@Service
public class RabbitMQService { 
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("fanout_queue_email"),
            exchange=@Exchange(value = "fanout_exchange",type = "fanout")
    ))
    public void subConsumerEmail(Message message){
        //當(dāng)隊(duì)列中有內(nèi)容是方法會(huì)自動(dòng)執(zhí)行   推薦Object來(lái)接收
        //官網(wǎng)推薦Message
        byte[] body = message.getBody();//Message將數(shù)據(jù)存放在body中
        String msg = new String(body);
        System.out.println("郵件業(yè)務(wù)接收到消息:"+msg);
    }
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("fanout_queue_sms"),
            exchange=@Exchange(value = "fanout_exchange",type = "fanout")
    ))
    public void subConsumerSms(Message message){
        byte[] body = message.getBody();
        String msg = new String(body);
        System.out.println("短信業(yè)務(wù)接收到消息:"+msg);
    }
}

提前將交換機(jī)和隊(duì)列刪除,然后運(yùn)行,就會(huì)發(fā)現(xiàn)會(huì)在啟動(dòng)時(shí)會(huì)自動(dòng)生成交換機(jī)和隊(duì)列,測(cè)試也不會(huì)有影響

到此這篇關(guān)于SpringBoot實(shí)現(xiàn)RabbitMQ三種使用方式的文章就介紹到這了,更多相關(guān)SpringBoot RabbitMQ使用內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: