Spring?Boot+RabbitMQ?通過(guò)fanout模式實(shí)現(xiàn)消息接收功能(支持消費(fèi)者多實(shí)例部署)

更新時(shí)間：2022年03月03日 08:57:23 作者：SKevin

這篇文章主要介紹了Spring?Boot+RabbitMQ?通過(guò)fanout模式實(shí)現(xiàn)消息接收（支持消費(fèi)者多實(shí)例部署）,本文通過(guò)案例場(chǎng)景分析給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下

　　本文章適用的場(chǎng)景：同一條消息可以被多個(gè)消費(fèi)者同時(shí)消費(fèi)。注意：當(dāng)消費(fèi)者多實(shí)例部署時(shí)，會(huì)輪詢消費(fèi)消息。網(wǎng)上有大量的的案例展示：P生產(chǎn)一條消息，消費(fèi)者服務(wù)C中建立Q1和Q2兩個(gè)隊(duì)列共同消費(fèi)。但極少的材料展示：P生產(chǎn)一條消息后M1，消費(fèi)者C1和C2可以同時(shí)消費(fèi)M1，如下圖所示。案例基于Spring Boot以及RabbitMQ的“fanout”類型exchange。已經(jīng)實(shí)測(cè)可放心使用。

1、引入基本依賴，項(xiàng)目不同請(qǐng)您按自己的情況引入合適的依賴

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

2、rabbitmq密碼、用戶名等配置

　　內(nèi)容過(guò)于基礎(chǔ)，請(qǐng)網(wǎng)絡(luò)上自行搜索相關(guān)內(nèi)容。

3、Exchange配置

@Bean
    public FanoutExchange myFanoutExchange() {
　 //“my.fanout”僅用于演示，請(qǐng)按自己的項(xiàng)目情況選擇合適的、有意義的命名
        return new FanoutExchange("my.fanout", true, false);
    }
    @Bean
    public Queue anonymousQueue() {
        //注意：必須使用匿名隊(duì)列才能達(dá)到本文目標(biāo)，此匿名隊(duì)列實(shí)例的名稱在消費(fèi)端會(huì)被使用
        return new AnonymousQueue();
    public Binding bindAnonymousQueue() {
        return BindingBuilder.bind(anonymousQueue()).to(myFanoutExchange());

4、發(fā)送端代碼

//構(gòu)建回調(diào)返回的數(shù)據(jù)
        CorrelationData correlationData = new CorrelationData(UUIDUtils.getUUID());

        Message msg = MessageBuilder.withBody(message.getBytes())
                .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
                .setCorrelationId(correlationData.getId())
                .build();
        try {
            rabbitTemplate.convertAndSend(exchangeName, null, msg, correlationData);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }

注意：“fanout"模式直接將消息發(fā)送到“exchange”，routing key應(yīng)為null或“”，如果您的代碼只需要發(fā)送消息而無(wú)接收，請(qǐng)忽略第3小節(jié)“Exchange配置”中所示代碼

5、消費(fèi)者代碼

@Service
public class MyListener {
    @RabbitListener(queues = "#{anonymousQueue.name}")
    public void receive(String message, Message message1, Channel channel) {
        try {
            //your code
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            long deliverTag = message1.getMessageProperties().getDeliveryTag();
            channel.basicAck(deliverTag, false);
        }
    }
}

　　代碼“@RabbitListener…”中的“#{anonymousQueue.name}”中的“anonymousQueue”對(duì)應(yīng)第3小節(jié)“Exchange配置”中所聲明的匿名隊(duì)列實(shí)例。

到此這篇關(guān)于Spring Boot+RabbitMQ 通過(guò)fanout模式實(shí)現(xiàn)消息接收（支持消費(fèi)者多實(shí)例部署）的文章就介紹到這了,更多相關(guān)Spring Boot RabbitMQ實(shí)現(xiàn)消息接收內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: