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

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

本篇文章主要講述Spring Boot與RabbitMQ的整合。因?yàn)槲覀児镜脑品?wù)用到了RabbitMQ 技術(shù)，之前都是自己封裝，正好我們也正在往SpringBoot轉(zhuǎn)變，這個(gè)技術(shù)正好用到，看來代碼又要重構(gòu)咯。

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

導(dǎo)包：

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

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 【shizhiyuan@usr.cn】
 * @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)用才能進(jìn)行消息的回調(diào)  必須要設(shè)置
        connectionFactory.setPublisherConfirms(true);
        return connectionFactory;
    }

  

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

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

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

調(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 【shizhiyuan@usr.cn】
 * @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:交換機(jī)名稱,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("消息成功消費(fèi)");
        } else {
            System.out.println("消息消費(fèi)失敗:" + cause);
        }
    }
}

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

還是在AmqpConfig.class里面

步驟就是

聲明交換機(jī)

聲明隊(duì)列

綁定RoutingKey

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

     * 

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

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

    @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è)置確認(rèn)模式手工確認(rèn)
        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));
                //確認(rèn)消息成功消費(fèi)
                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 【shizhiyuan@usr.cn】
 * @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)用才能進(jìn)行消息的回調(diào)  必須要設(shè)置
        connectionFactory.setPublisherConfirms(true);
        return connectionFactory;
    }

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


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

     * 

     * FanoutExchange: 將消息分發(fā)到所有的綁定隊(duì)列，無routingkey的概念
     * HeadersExchange ：通過添加屬性key-value匹配
     * DirectExchange:按照routingkey分發(fā)到指定隊(duì)列
     * 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è)置確認(rèn)模式手工確認(rèn)
        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));
                //確認(rèn)消息成功消費(fèi)
                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]
消息成功消費(fèi)

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

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