RabbitMQ發送和接收消息的幾種方式

作者：程序猿羊 2023-11-08 07:51:11

channel.basicQos(0, 1, false)：0表示對消息的大小無限制，1表示每次只允許消費一條，false表示該限制不作用于channel。同時，我們采用手工ACK的方式，因為我們配置文件配置了 spring.rabbitmq.listener.simple.acknowledge-mode=manual。

一、發送消息的幾種方式

1.1、默認交換機和routingKey----(個人不推薦使用)

使用默認的交換機exchange或routingKey。

圖片

調用方法：

圖片

1.2、使用指定routingKey的方式發送(默認的交換機)

使用默認的交換機，routingKey必須為quenue隊列的名稱。

調用方法：

圖片

案例：

/**
 * @Author yangyalin
 * @Description 測試發送消息(直接使用隊列發送,使用默認的交換機) routingKey：即為對列的名稱即可
 **/
public void testSendMsg(String message){
    rabbitTemplate.convertAndSend(RabbitMQConvertConfig.TEST_QUEUE,message);
}

1.3、指定交換機和routingKey的方式發送

使用指定的交換機，若綁定routingKey,必須使用指定的模式；若沒有綁定，可設置為""。

調用方法：

圖片

案例：

public void sendDecreStockMessage(DecreStockFromRabbit decreStockFromRabbit){
    CorrelationData correlationData = new CorrelationData();
    correlationData.setId(decreStockFromRabbit.getMessageId());
  /**
   * exchange：交換機  routingKey：路由鍵  message：消息體內容   correlationData：消息唯一ID
  **/
    rabbitTemplate.convertAndSend(RabbitMQConvertConfig.ORDER_EXCHANGE,
        RabbitMQConvertConfig.ORDER_ROUTINGKEY, decreStockFromRabbit,correlationData);
}
或：
rabbitTemplate.convertAndSend("test-exchange","",message);

二、接收消息的幾種方式

2.1、默認交換機，提前創建好隊列(TestDirectQueue)

/**
     * 功能描述:當消費同一個隊列的時候，可通過設置實現能則多勞，
     * 消息輪詢方式訂閱
     * @MethodName: process11
     * @MethodParam: [testMessage]
     * @Return: void
     * @Author: yyalin
     * @CreateDate: 2022/4/9 17:10
     */
    @RabbitListener(queues = "TestDirectQueue")   //監聽的隊列名稱 TestDirectQueue
    public void process11(Map testMessage) throws InterruptedException {
        log.info("消費者收到消息222:" + testMessage.toString());
        Thread.sleep(200);
    }

2.2、默認交換機，自動創建隊列(TEST_QUEUE2)

@RabbitListener(queuesToDeclare=@Queue(TopicExchangeConfig.TEST_QUEUE2))
@RabbitHandler
public void receiveTestMsg2(@Payload String str) throws Exception{
    log.info("開始接收消息。。。。。");
    log.info("接收到的消息："+str);
}

2.3、自動創建交換機和隊列----(個人推薦)

自動創建且交換機和隊列綁定,key可指定也可不指定(默認為隊列名稱)。

/******************方案二：使用注解的方式綁定隊列在交換機上*******************/
    @RabbitListener(bindings = @QueueBinding(value=@Queue(name="directQueue"),
            exchange=@Exchange(name="directExchange",type = ExchangeTypes.DIRECT),
            key={"red", "blue"}
    ))    //監聽的隊列名稱 TestDirectQueue
    public void directConsumer(String message) {
        log.info("消費者收到direct消息555  : " + message);
    }


   @RabbitListener(bindings = @QueueBinding(value=@Queue(name="topicQueue2"),
            exchange=@Exchange(name="topicExchange",type = ExchangeTypes.TOPIC,ignoreDeclarationExceptions = "true"),
            key="#.new"
    ))
    public void topicConsumer2(String message) {
        log.info("消費者收到topic消息888  : " + message);
    }

備注：ignoreDeclarationExceptions = "true" : 即使配置出現了錯誤也不至于整個應用程序都啟動失敗的情況。

1、channel.basicQos(0, 1, false)：0表示對消息的大小無限制，1表示每次只允許消費一條，false表示該限制不作用于channel。

同時，我們采用手工ACK的方式，因為我們配置文件配置了 spring.rabbitmq.listener.simple.acknowledge-mode=manual：

2、channel.basicAck(deliveryTag, false)：deliveryTag表示處理的消息條數（一般為1），從heaers中取，false表示不批量ack。

/**
     * 功能描述: 消費端加上手動確認消息被接收
     * @MethodName: process
     * @MethodParam: [message]
     * @Return: void
     * @Author: yyalin
     * @CreateDate: 2022/4/18 19:10
     */
    @RabbitListener(queues = "TestDirectQueue3")    //監聽的隊列名稱 TestDirectQueue
    public void process(String message, Channel channel) throws IOException {
        log.info("DirectReceiver消費者收到消息1  : " + message);
        long msgId=1111L; //消息ID
        try {
            //手動確認消息已消費
            channel.basicAck(msgId,false);
        } catch (IOException e) {
            //把消息失敗的消息重新放入到隊列
            channel.basicNack(msgId,false,true);
            e.printStackTrace();
        }
    }

責任編輯：武曉燕來源：程序猿小楊

RabbitMQ 接收消息