Springboot怎麼整合RabbitMQ訊息佇列

生產者工程

POM依賴

可以在建立工程時直接選擇新增依賴。

application檔案

因為rabbitmq有預設位址及使用者訊息，所以如果是本地rabbitmq可以不需要進行配置。

RabbitMQ設定檔：

在使用相關交換器及佇列時，我們需要實作聲明交換器及佇列，如果沒有對應訊息，啟動專案會失敗。所以在使用springboot整合rabbitmq時，我們可以透過設定檔來進行交換器、佇列的宣告及二者之間的關係綁定。由於目前在示範Fanout模式，所以使用FanoutExchange來宣告交換機，其他模式則使用對應的TopicExchange，DirectExchange來宣告。

@Configuration
public class RabbitMQConfiguration {

//声明fanout模式的交换机
@Bean
public FanoutExchange fanoutExchange() {
    return new FanoutExchange("fanout_order_exchange", true, false);
}

//声明队列
@Bean
public Queue smsQueue() {
    return new Queue("sms.fanout.queue", true);
}

@Bean
public Queue emailQueue() {
    return new Queue("email.fanout.queue", true);
}

@Bean
public Queue duanxinQueue() {
    return new Queue("duanxin.fanout.queue", true);
}
//绑定

@Bean
public Binding smsBinding() {
    return BindingBuilder.bind(smsQueue()).to(fanoutExchange());
}

@Bean
public Binding emailBinding() {
    return BindingBuilder.bind(emailQueue()).to(fanoutExchange());
}

@Bean
public Binding duanxinBinding() {
    return BindingBuilder.bind(duanxinQueue()).to(fanoutExchange());
}
}

生產者業務代碼

這部分程式碼就簡單的透過呼叫rabbitTemplate來進行訊息的分發。 @Service public class OrderService {

@Autowired
private RabbitTemplate rabbitTemplate;

public void makeOrder() {
    // 保存订单
    String orderId = UUID.randomUUID().toString();
    System.out.println("下单成功：" + orderId);
    // 通过MQ完成消息的分发
    // 参数1：交换机 ；参数2：路由key/队列名；参数3：消息内容
    String exchangeName = "fanout_order_exchange";
    rabbitTemplate.convertAndSend(exchangeName, "", orderId);
}
}

消費者：

#消費者工程和生產者工程類似，我們首先需要引入依賴，然後在application文件中進行相關的配置即可開始編寫程式碼。在消費者工程中我們也可以寫rabbitmq的設定檔來進行交換器及佇列的聲明。建議在消費端編寫設定文件，因為消費端是先啟動的工程，如果交換器和佇列未建立會導致工程啟動失敗。訊息監聽

我們透過RabbitListener註解來監聽訊息佇列。要注意的是我們需要透過Component註解將該監聽交給spring管理，否則無法正常接收服務端的訊息。這邊只給一個email的訊息監聽，上文生產者聲明的duanxin,sms佇列可以自行創建，只需要修改佇列名稱即可。 @Service public class OrderService {

@RabbitListener(queues = {"email.fanout.queue"})
@Component
public class FanoutEmailService {
    @RabbitHandler
    public void receive(String message) {
        System.out.println("email fanout -----》接收到" + message);
    }
}

測試

先啟動消費者工程，然後在生產者工程中建立測試類別發送訊息即可。

@SpringBootTest class SpringbootOrderRabbitmqProducerApplicationTests {

@Autowired
private OrderService orderService;

@Test
void contextLoads() {
    orderService.makeOrder();
}
}

當發送訊息後，我們可以在控制台中發現消費者成功接受訊息。

Direct 模式

生產者

建立工程的步驟和前文相同。

設定檔

設定和上文基本上相同，由於該部分測試direct模式，所以需要使用DirectExchange建立交換器。要注意的是該類別中的方法名稱不能和上文rabbitmq的設定檔中的方法名稱相同，因為我們使用bean註解將其交給spring管理，如果名字相同，則會啟動專案失敗。

@Configuration
public class DirectRabbitMQConfiguration {

//声明direct模式的交换机
@Bean
public DirectExchange directExchange() {
    return new DirectExchange("direct_order_exchange", true, false);
}

//声明队列
@Bean
public Queue smsDirectQueue() {
    return new Queue("sms.direct.queue", true);
}

@Bean
public Queue emailDirectQueue() {
    return new Queue("email.direct.queue", true);
}

@Bean
public Queue duanxinDirectQueue() {
    return new Queue("duanxin.direct.queue", true);
}
//绑定

@Bean
public Binding smsDirectBinding() {
    return BindingBuilder.bind(smsDirectQueue()).to(directExchange()).with("sms");
}

@Bean
public Binding emailDirectBinding() {
    return BindingBuilder.bind(emailDirectQueue()).to(directExchange()).with("email");
}

@Bean
public Binding duanxinDirectBinding() {
    return BindingBuilder.bind(duanxinDirectQueue()).to(directExchange()).with("duanxin");
}
}

業務代碼

@Service
public class OrderService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void makeOrderDirect() {
        // 保存订单
        String orderId = UUID.randomUUID().toString();
        System.out.println("下单成功：" + orderId);
        String exchangeName = "direct_order_exchange";
        rabbitTemplate.convertAndSend(exchangeName, "sms", orderId);
        rabbitTemplate.convertAndSend(exchangeName, "email", orderId);
    }

}

消費者

訊息監聽

和上文相同，只要注意佇列名稱即可。

@RabbitListener(queues = {"email.direct.queue"})
@Component
public class DirectEmailService {
    @RabbitHandler
    public void receive(String message) {
        System.out.println("email direct -----》接收到" + message);
    }
}

Topic 模式

上文中個模式都是透過設定檔來宣告交換機，佇列及綁定二者之間的關係；其實我們也可以透過註解的方式來宣告交換機及註解。

生產者

由於使用註解方式聲明，所以我們不需要建立設定文件，直接編寫業務程式碼即可。測試的時候我們只需修改路由名稱即可，具體如何修改，請前往文章開頭連結查看各模式是如何使用的。

@Service
public class OrderService {

@Autowired
private RabbitTemplate rabbitTemplate;

public void makeOrderTopic() {
    // 保存订单
    String orderId = UUID.randomUUID().toString();
    System.out.println("下单成功：" + orderId);
    String exchangeName = "topic_order_exchange";
    String routingKey = "com.email";
    rabbitTemplate.convertAndSend(exchangeName, routingKey, orderId);
}
}

消費者

程式碼和上文基本上相同，差別在於我們直接在RabbitListener註解中將佇列和交換器進行綁定。需要注意的是各參數中都是使用字串。 value對應的是隊列，對應的參數分別是隊列名、持久化、自動刪除。 exchange對應的交換機，對應的參數分別是交換器名以及交換器類型。 key對應的是路由名。

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "email.topic.queue",durable = "true",autoDelete = "false"),
        exchange = @Exchange(value = "topic_order_exchange",type = ExchangeTypes.TOPIC),
        key = "*.email.#"
))
@Component
public class TopicEmailService {
    @RabbitHandler
    public void receive(String message) {
        System.out.println("email topic -----》接收到" + message);
    }
}

以上是Springboot怎麼整合RabbitMQ訊息佇列的詳細內容。更多資訊請關注PHP中文網其他相關文章！