基於rabbitmq的Spring-amqp基本使用

目錄

• 1. 依賴和配置
  • 添加AMQP的啟動器：
  • 在application.yml中添加RabbitMQ地址：
• 2. 監聽者
• 3. AmqpTemplate
• 4. 測試代碼

Spring-amqp是對AMQP協議的抽象實現，而spring-rabbit 是對協議的具體實現，也是目前的唯一實現。底層使用的就是RabbitMQ。

1. 依賴和配置

添加AMQP的啟動器：

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

在application.yml中添加RabbitMQ地址：

spring:
  rabbitmq:
    host: 192.168.0.22
    username: leyou
    password: leyou
    virtual-host: /leyou

2. 監聽者

在SpringAmqp中，對消息的消費者進行了封裝和抽象，一個普通的JavaBean中的普通方法，只要通過簡單的注解，就可以成為一個消費者。

import org.springframework.amqp.core.ExchangeTypes;
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.Component;

/**
 * @author john
 * @date 2019/12/12 - 8:04
 */
@Component
public class Listener {

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "spring.test.queue", durable = "true"),
            exchange = @Exchange(
                    value = "spring.test.exchange",
                    ignoreDeclarationExceptions = "true",
                    type = ExchangeTypes.TOPIC
            ),
            key = {"#.#"}))
    public void listen(String msg){
        System.out.println("接收到消息：" + msg);
    }
}

• @Componet：類上的注解，注冊到Spring容器
• @RabbitListener：方法上的注解，聲明這個方法是一個消費者方法，需要指定下面的屬性：
  • bindings：指定綁定關系，可以有多個。值是@QueueBinding的數組。@QueueBinding包含下面屬性：
    • value：這個消費者關聯的隊列。值是@Queue，代表一個隊列
    • exchange：隊列所綁定的交換機，值是@Exchange類型
    • key：隊列和交換機綁定的RoutingKey

類似listen這樣的方法在一個類中可以寫多個，就代表多個消費者。

3. AmqpTemplate

Spring最擅長的事情就是封裝，把他人的框架進行封裝和整合。

Spring為AMQP提供了統一的消息處理模板：AmqpTemplate，非常方便的發送消息，其發送方法：

紅框圈起來的是比較常用的3個方法，分別是：

• 指定交換機、RoutingKey和消息體
• 指定消息
• 指定RoutingKey和消息，會向默認的交換機發送消息

4. 測試代碼

@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class MqDemo {

    @Autowired
    private AmqpTemplate amqpTemplate;

    @Test
    public void testSend() throws InterruptedException {
        String msg = "hello, Spring boot amqp";
        this.amqpTemplate.convertAndSend("spring.test.exchange","a.b", msg);
        // 等待10秒后再結束
        Thread.sleep(10000);
    }
}

運行后查看日志：