RabbitMQ在springboot中實現(xiàn)短信服務(wù)的異步通信方式

更新時間：2026年05月19日 16:46:09 作者：胡思亂想罷了~

RabbitMQ是一個開源消息中間件,適用于異步通信、應(yīng)用解耦、可靠消息傳遞、靈活路由、延遲消息處理和分布式系統(tǒng)調(diào)等場景,SpringBoot使用需要新建隊列、交換機并配置綁定,文章通過短信驗證碼服務(wù)為例,演示了RabbitMQ在項目中的應(yīng)用場景和配置方法

springboot中RabbitMQ基本使用

RabbitMQ是一個開源的消息中間件

具有以下用途和優(yōu)點：

異步通信：RabbitMQ可以作為應(yīng)用程序之間進(jìn)行異步通信的媒介，生產(chǎn)者可以將消息發(fā)送到RabbitMQ，而消費者可以從中獲取消息并進(jìn)行處理。這種異步通信可以實現(xiàn)系統(tǒng)各部分的解耦，提高系統(tǒng)的可伸縮性和性能。
應(yīng)用解耦：通過使用RabbitMQ，不同的應(yīng)用程序可以通過發(fā)送和接收消息來解耦，并減少應(yīng)用之間的依賴。這樣可以提高系統(tǒng)的靈活性，使系統(tǒng)更易于維護(hù)和擴展。
可靠消息傳遞：RabbitMQ確保消息傳遞的可靠性。它提供持久化和消息確認(rèn)機制，即使在消息發(fā)送或消費者出現(xiàn)故障的情況下，也能保證消息的不丟失和不重復(fù)。
靈活的消息路由和交換：RabbitMQ支持多種消息路由模式和交換機類型，如直連、扇形、主題和頭部交換機。這使得可以根據(jù)消息的內(nèi)容和標(biāo)簽將消息路由到不同的隊列中，滿足各種復(fù)雜的應(yīng)用場景。
延遲消息處理：RabbitMQ支持延遲消息的發(fā)送和處理，可以用于處理定時任務(wù)、延遲通知和調(diào)度等需求。
分布式系統(tǒng)中的協(xié)調(diào)：RabbitMQ在分布式系統(tǒng)中扮演著關(guān)鍵的角色，可以用于實現(xiàn)分布式系統(tǒng)的任務(wù)分發(fā)、事件發(fā)布訂閱、分布式鎖等。它提供了一種簡單而強大的方式來處理分布式系統(tǒng)間的通信和協(xié)調(diào)機制。

RabbitMQ是一個功能強大、可靠性高且易于使用的消息中間件。它在異步通信、應(yīng)用解耦、可靠消息傳遞、靈活的消息路由和交換、延遲消息處理以及分布式系統(tǒng)中的協(xié)調(diào)等方面具有廣泛應(yīng)用。

在SpringbBoot在使用首先需要在rabbitmq管理界面新建隊列和交換機。

RabbitMQ支持多種類型的交換機，用于定義消息的路由規(guī)則。

以下是幾種常見的交換機類型：

直連交換機（Direct Exchange）：根據(jù)消息的路由鍵（Routing Key）精確地將消息路由到匹配的隊列。例如，通過將消息的路由鍵設(shè)置為特定的關(guān)鍵詞，可以將消息發(fā)送到對應(yīng)的隊列，以便消費者進(jìn)行處理。
扇形交換機（Fanout Exchange）：將消息廣播到所有與交換機綁定的隊列。無論消息的路由鍵是什么，扇形交換機都會將消息發(fā)送到所有綁定的隊列中。適用于多個消費者需要處理相同消息的場景。
主題交換機（Topic Exchange）：根據(jù)路由鍵的模式進(jìn)行匹配和路由?？梢允褂猛ㄅ浞?ldquo;*”匹配一個單詞，使用“#”匹配零個或多個單詞。主題交換機可以滿足更復(fù)雜的路由需求，可以將消息路由到多個隊列。
頭部交換機（Headers Exchange）：根據(jù)消息的頭部屬性進(jìn)行匹配和路由。消息發(fā)布者在發(fā)送消息時可以在消息的頭部添加自定義的屬性，交換機將根據(jù)這些屬性進(jìn)行匹配和路由。

不同類型的交換機適用于不同的應(yīng)用場景，開發(fā)者可以根據(jù)具體需求選擇適合的交換機類型來進(jìn)行消息路由和分發(fā)。

在項目中引入依賴

<!-- RabbitMQ -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

yml文件配置

rabbitmq的用戶名和密碼都為默認(rèn)

spring:
  datasource:
    driver-class-name: com.mysql.cj.jdbc.Driver
    url: jdbc:mysql://localhost:3306/market?useSSl=true&useUnicode=true$characterEncoding=utf-8&serverTimezone=GMT%2B8
    username: root
    password: 010409
  rabbitmq:
      host: 127.0.0.1
      port: 5672
      username: guest
      password: guest

交換機與隊列綁定配置

@Configuration
public class RabbitMQConfig {
    private static final String MARKET = "market";
    private static final String ORDER = "order";
    private static final String EXCHANGE = "market.exChanges";
    @Bean
    public Queue queue01(){
        return new Queue(MARKET);
    }
    @Bean
    public Queue queue02(){
        return new Queue(ORDER);
    }
    @Bean
    public DirectExchange fanoutExchange(){
        return new DirectExchange(EXCHANGE);
    }
    // 綁定交換機和隊列
    @Bean
    public Binding binding01(){
        return BindingBuilder.bind(queue01()).to(fanoutExchange()).with("market");
    }
    @Bean
    public Binding binding02(){
        return BindingBuilder.bind(queue02()).to(fanoutExchange()).with("order");
    }
}

應(yīng)用場景

短信驗證碼服務(wù)，這里我使用的是阿里云短信服務(wù)SMS，將短信驗證碼與前端輸入的手機號存入隊列中，將驗證碼存入redis中，方便后續(xù)注冊時驗證碼驗證，最后監(jiān)聽隊列拿到信息后發(fā)送短信。

SMS依賴與配置，需要將配置修改為自己的

<!--  阿里云短信依賴  -->
<dependency>
    <groupId>com.aliyun</groupId>
    <artifactId>aliyun-java-sdk-core</artifactId>
    <version>4.5.16</version>
</dependency>
<dependency>
    <groupId>com.aliyun</groupId>
    <artifactId>aliyun-java-sdk-dysmsapi</artifactId>
    <version>2.1.0</version>
</dependency>

@Service
public class SmsService {
    public boolean send(Map<String, Object> param, String phone) {

        if(StringUtils.isEmpty(phone)) return false;
        //default 地域節(jié)點，默認(rèn)就好  后面是 阿里云的 id和秘鑰（這里記得去阿里云復(fù)制自己的id和秘鑰哦）
        DefaultProfile profile = DefaultProfile.getProfile("default",  "LTAI5tNtEzgs7etoRVhWQuXz", "X8odaVou1eSwT0boKEgvYxPSUtWafL");
        IAcsClient client = new DefaultAcsClient(profile);

        CommonRequest request = new CommonRequest();

        request.setProtocol(ProtocolType.HTTPS);
        request.setMethod(MethodType.POST);
        request.setDomain("dysmsapi.aliyuncs.com");
        request.setVersion("2017-05-25");
        request.setAction("SendSms");

        request.putQueryParameter("PhoneNumbers", phone);  //手機號
        request.putQueryParameter("SignName", "阿里云短信測試");    //申請阿里云 簽名名稱（暫時用阿里云測試的，自己還不能注冊簽名）
        request.putQueryParameter("TemplateCode",  "SMS_154950909"); //申請阿里云 模板code（用的也是阿里云測試的）
        request.putQueryParameter("TemplateParam", JSONObject.toJSONString(param));

        try {
            CommonResponse response = client.getCommonResponse(request);
//            System.out.println(response.getData());
            return response.getHttpResponse().isSuccess();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }
}

發(fā)送短信的接口：

@RequestMapping("/sendCode")
public Result sendCode(String phone){
    return UserService.sendCode(phone);
}

實現(xiàn)：

   @Override
    public Result sendCode(String phone) {
        // 1.校驗手機號
        String regex = "^1[3456789]\\d{9}$";

        if (!phone.matches(regex)) {
            // 2.如果不符合，返回錯誤信息
            return new Result(100,"手機號格式錯誤！");
        }

//      3. 生成驗證碼
        String code = "";
        for (int i = 0; i < 6; i++) {
            int var = new Random().nextInt(10);
            code = code + var;
        }

        Map<String,Object> param = new HashMap<>();
        param.put("code", code);

        // 4.保存驗證碼到 redis
        redisCache.setCacheObject(phone,code);

        Map<String,String> map=new HashMap();
        map.put("phone", phone);
        map.put("code", code);
        // 5.保存驗證碼和手機號到隊列
        rabbitTemplate.convertAndSend("market.exChanges","market",map);

        return new Result(200, "發(fā)送成功");
    }

監(jiān)聽隊列并發(fā)送短信。

@RabbitListener(queues = "market")
@RabbitHandler
public void executeSPC(Map<String, String> map, Channel channel, @Headers Map<String, Object> headers) throws IOException {
    try {
        Map<String,Object> param = new HashMap<>();
        param.put("code", map.get("code"));
        smsService.send(param, map.get("phone"));

        // 手動發(fā)送ACK，確認(rèn)消息已接收
        Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
        channel.basicAck(deliveryTag, false);
    } catch (Exception e) {
        // 處理異常情況，根據(jù)業(yè)務(wù)需要選擇是否重新入列或者拒絕消息等操作
        Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
        channel.basicNack(deliveryTag, false, true);
    }
}