Spring boot整合RabbitMQ

snifes 發(fā)布于2019-08-15 14:19 / 2698人閱讀

摘要：如果你注定要成為厲害的人那問題的答案就深藏在你的血脈里。本篇文章主要講述與的整合。有想了解重構(gòu)的朋友，我之前也有對重構(gòu)一書的解讀，出門左轉(zhuǎn)就能看到。

如果你注定要成為厲害的人, 那問題的答案就深藏在你的血脈里。

本篇文章主要講述Spring Boot與RabbitMQ的整合。因為我們公司的云服務(wù)用到了RabbitMQ 技術(shù)，之前都是自己封裝，正好我們也正在往SpringBoot轉(zhuǎn)變，這個技術(shù)正好用到，看來代碼又要重構(gòu)咯。

有想了解重構(gòu)的朋友，我之前也有對《重構(gòu)》一書的解讀，出門左轉(zhuǎn)就能看到。

導(dǎo)包：

        
            org.springframework.boot
            spring-boot-starter-amqp

消息生產(chǎn)者

ConnectionFactory配置
創(chuàng)建AmqpConfig文件AmqpConfig.java(后期的配置都在該文件中)

package cn.usr.springbootrabbitmq;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;

/**
 * @program: Learn-SpringBootRabbitmq
 * @author: Rock 【[email protected]】
 * @Date: 2018/2/23 0023
 */
@Configuration
public class AmqpConfig {
    public static final String EXCHANGE = "spring-boot-exchange2";
    public static final String ROUTINGKEY = "spring-boot-routingKey2";


    @Bean
    public CachingConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setAddresses("127.0.0.1");
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setVirtualHost("/");
        // 這里需要顯示調(diào)用才能進行消息的回調(diào)  必須要設(shè)置
        connectionFactory.setPublisherConfirms(true);
        return connectionFactory;
    }

RabbitTemplate

    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate template = new RabbitTemplate(connectionFactory());
        return template;
    }

這里設(shè)置為原型，具體的原因在后面會講到，在發(fā)送消息時通過調(diào)用RabbitTemplate中的如下方法：
一會調(diào)用的時候用：

 public void convertAndSend(String exchange, String routingKey, Object object, CorrelationData correlationData)

Producer

調(diào)用啦：

package cn.usr.springbootrabbitmq;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.UUID;

/**
 * @program: Learn-SpringBootRabbitmq
 * @author: Rock 【[email protected]】
 * @Date: 2018/2/23 0023
 */
@Component
public class Producer implements RabbitTemplate.ConfirmCallback {
    private RabbitTemplate rabbitTemplate;


    /**
     * 構(gòu)造方法注入
     */
    @Autowired
    public Producer(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
        //這是是設(shè)置回調(diào)能收到發(fā)送到響應(yīng)，confirm()在下面解釋
        rabbitTemplate.setConfirmCallback(this);
    }

    public void sendMsg(String content) {
        CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
        //convertAndSend(exchange:交換機名稱,routingKey:路由關(guān)鍵字,object:發(fā)送的消息內(nèi)容,correlationData:消息ID)
        rabbitTemplate.convertAndSend(AmqpConfig.EXCHANGE, AmqpConfig.ROUTINGKEY, content, correlationId);
    }

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        System.out.println(" 回調(diào)id:" + correlationData);
        if (ack) {
            System.out.println("消息成功消費");
        } else {
            System.out.println("消息消費失敗:" + cause);
        }
    }
}

如果需要在生產(chǎn)者需要消息發(fā)送后的回調(diào)，需要對rabbitTemplate設(shè)置ConfirmCallback對象，由于不同的生產(chǎn)者需要對應(yīng)不同的ConfirmCallback，如果rabbitTemplate設(shè)置為單例bean，則所有的rabbitTemplate實際的ConfirmCallback為最后一次申明的ConfirmCallback。

消息消費者

還是在AmqpConfig.class里面

步驟就是

聲明交換機

聲明隊列

綁定RoutingKey

/**
     * 針對消費者配置
     * 1. 設(shè)置交換機類型
     * 2. 將隊列綁定到交換機
     * 
     * 

     * FanoutExchange: 將消息分發(fā)到所有的綁定隊列，無routingkey的概念
     * HeadersExchange ：通過添加屬性key-value匹配
     * DirectExchange:按照routingkey分發(fā)到指定隊列
     * TopicExchange:多關(guān)鍵字匹配
     */
    @Bean
    public DirectExchange defaultExchange() {
        return new DirectExchange(EXCHANGE);
    }

    @Bean
    public Queue queue() {
        return new Queue("spring-boot-queue", true);//隊列持久 
    }

    @Bean
    public Binding binding() {
        return BindingBuilder.bind(queue()).to(defaultExchange()).with(AmqpConfig.ROUTINGKEY);
    }


    @Bean
    public SimpleMessageListenerContainer messageContainer() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
        container.setQueues(queue());
        container.setExposeListenerChannel(true);
        container.setMaxConcurrentConsumers(1);
        container.setConcurrentConsumers(1);
        // 設(shè)置確認模式手工確認
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setMessageListener(new ChannelAwareMessageListener() {
            @Override
            public void onMessage(Message message, Channel channel) throws Exception {
                byte[] body = message.getBody();
                System.out.println("receive msg : " + new String(body));
                //確認消息成功消費
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            }
        });
        return container;
    }

下面是完整的配置：

package cn.usr.springbootrabbitmq;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;

/**
 * @program: Learn-SpringBootRabbitmq
 * @author: Rock 【[email protected]】
 * @Date: 2018/2/23 0023
 */
@Configuration
public class AmqpConfig {
    public static final String EXCHANGE = "spring-boot-exchange2";
    public static final String ROUTINGKEY = "spring-boot-routingKey2";


    @Bean
    public CachingConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setAddresses("127.0.0.1");
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setVirtualHost("/");
        // 這里需要顯示調(diào)用才能進行消息的回調(diào)  必須要設(shè)置
        connectionFactory.setPublisherConfirms(true);
        return connectionFactory;
    }

    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate template = new RabbitTemplate(connectionFactory());
        return template;
    }


    /**
     * 針對消費者配置
     * 1. 設(shè)置交換機類型
     * 2. 將隊列綁定到交換機
     * 
     * 

     * FanoutExchange: 將消息分發(fā)到所有的綁定隊列，無routingkey的概念
     * HeadersExchange ：通過添加屬性key-value匹配
     * DirectExchange:按照routingkey分發(fā)到指定隊列
     * TopicExchange:多關(guān)鍵字匹配
     */
    @Bean
    public DirectExchange defaultExchange() {
        return new DirectExchange(EXCHANGE);
    }

    @Bean
    public Queue queue() {
        return new Queue("spring-boot-queue", true);
    }

    @Bean
    public Binding binding() {
        return BindingBuilder.bind(queue()).to(defaultExchange()).with(AmqpConfig.ROUTINGKEY);
    }


    @Bean
    public SimpleMessageListenerContainer messageContainer() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
        container.setQueues(queue());
        container.setExposeListenerChannel(true);
        container.setMaxConcurrentConsumers(1);
        container.setConcurrentConsumers(1);
        // 設(shè)置確認模式手工確認
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setMessageListener(new ChannelAwareMessageListener() {
            @Override
            public void onMessage(Message message, Channel channel) throws Exception {
                byte[] body = message.getBody();
                System.out.println("receive msg : " + new String(body));
                //確認消息成功消費
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            }
        });
        return container;
    }
}

到這里我就能完成SpringBoot整合RabbitMQ的數(shù)據(jù)收發(fā)了。

結(jié)果：

receive msg : ceshi-----?
 回調(diào)id:CorrelationData [id=dfe3b3d1-f5a3-42d9-a514-a73729e009d5]
消息成功消費

點贊收藏關(guān)注不迷路。么么噠

參考：http://blog.csdn.net/liaokail...

文章版權(quán)歸作者所有，未經(jīng)允許請勿轉(zhuǎn)載,若此文章存在違規(guī)行為，您可以聯(lián)系管理員刪除。

轉(zhuǎn)載請注明本文地址：http://systransis.cn/yun/68539.html

發(fā)表評論

登陸后可評論

0條評論

snifes

男|高級講師

我要關(guān)注我要私信

TA的文章

如何保證消息隊列的可靠性傳輸？

閱讀 3334·2021-11-22 12:04
垂直水平居中的方式總結(jié) +（使用場景）

閱讀 2718·2019-08-29 13:49
(a== 1 && a ==2 && a==3) ever eval

閱讀 491·2019-08-26 13:45
html+css+js+Hbuilder開發(fā)一款安卓APP，根本不用學(xué)Android開發(fā)！

閱讀 2249·2019-08-26 11:56
每周一點canvas動畫——《支付寶價格拖動選擇》

閱讀 1007·2019-08-26 11:43
簡單HTML5移動端（客戶端）圖片上傳

閱讀 601·2019-08-26 10:45
前端開發(fā)面試題鏈接

閱讀 1275·2019-08-23 16:48
小程序顯示彈窗時禁止下層的內(nèi)容滾動

閱讀 2164·2019-08-23 16:07

成人国产在线小视频_日韩寡妇人妻调教在线播放_色成人www永久在线观看_2018国产精品久久_亚洲欧美高清在线30p_亚洲少妇综合一区_黄色在线播放国产_亚洲另类技巧小说校园_国产主播xx日韩_a级毛片在线免费

資訊專欄INFORMATION COLUMN

上云采購季！| 2核2G4M爆款云服務(wù)器低至59元/年，更有多臺、長期優(yōu)惠，快來選購！

Spring boot整合RabbitMQ

相關(guān)文章

SpringBoot RabbitMQ 整合使用

***SpringBoot非官方教程 | 第十五篇：Springboot整合RabbitMQ***

SpringBoot ActiveMQ 整合使用

spring boot - 收藏集 - 掘金

**慕課網(wǎng)_《RabbitMQ消息中間件極速入門與實戰(zhàn)》學(xué)習(xí)總結(jié)**

發(fā)表評論

0條評論

snifes

男|高級講師

TA的文章

如何保證消息隊列的可靠性傳輸？

垂直水平居中的方式總結(jié) +（使用場景）

(a== 1 && a ==2 && a==3) ever eval

html+css+js+Hbuilder開發(fā)一款安卓APP，根本不用學(xué)Android開發(fā)！

每周一點canvas動畫——《支付寶價格拖動選擇》

簡單HTML5移動端（客戶端）圖片上傳

前端開發(fā)面試題鏈接

小程序顯示彈窗時禁止下層的內(nèi)容滾動

最新活動

資訊專欄INFORMATION COLUMN

上云采購季！| 2核2G4M爆款云服務(wù)器低至59元/年，更有多臺、長期優(yōu)惠，快來選購！

Spring boot整合RabbitMQ

相關(guān)文章

發(fā)表評論

0條評論

男|高級講師

TA的文章

最新活動

上云采購季！| 2核2G4M爆款云服務(wù)器低至59元/年，更有多臺、長期優(yōu)惠，快來選購！