springboot中使用rabbitMQ

1、引入依赖

compile 'org.springframework.boot:spring-boot-starter-amqp'

2、配置文件：

spring.rabbitmq.host=10.0.60.121

spring.rabbitmq.port=5672

spring.rabbitmq.username=admin

spring.rabbitmq.password=admin

spring.rabbitmq.virtual-host=/

3、配置代码

import org.springframework.amqp.core.AmqpAdmin;

import org.springframework.amqp.core.Binding;

import org.springframework.amqp.core.BindingBuilder;

import org.springframework.amqp.core.DirectExchange;

import org.springframework.amqp.core.FanoutExchange;

import org.springframework.amqp.core.Queue;

import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;

import org.springframework.amqp.rabbit.connection.ConnectionFactory;

import org.springframework.amqp.rabbit.core.RabbitAdmin;

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.beans.factory.annotation.Value;

import org.springframework.beans.factory.config.ConfigurableBeanFactory;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

import org.springframework.context.annotation.PropertySource;

import org.springframework.context.annotation.Scope;

import com.wow.common.mq.MqMetaConstant;

@Configuration

@PropertySource("classpath:/rabbitmq.properties")

public class RabbitMqConfiguration {

    @Value("${spring.rabbitmq.host}")private String springRabbitmqHost;

    @Value("${spring.rabbitmq.port}") private Integer springRabbitmqPort;

   @Value("${spring.rabbitmq.username}") private String springRabbitmqUsername;

    @Value("${spring.rabbitmq.password}") private String springRabbitmqPassword;

    @Value("${spring.rabbitmq.virtual-host}") private String springRabbitmqVirtualHost;

    @Bean public ConnectionFactory connectionFactory() {

        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();

        //防止信道缓存不够造成消息丢失，官方推荐100可完全避免此丢失消息情况

        connectionFactory.setChannelCacheSize(100);

        connectionFactory.setHost(springRabbitmqHost);        

         connectionFactory.setPort(springRabbitmqPort);

        connectionFactory.setUsername(springRabbitmqUsername);

        connectionFactory.setPassword(springRabbitmqPassword);

        connectionFactory.setVirtualHost(springRabbitmqVirtualHost);

        connectionFactory.setChannelCacheSize(100);

        // 必须要设置 connectionFactory.setPublisherConfirms(true);

        return connectionFactory;

    }

        /** *

        * 用于恢复交换机，队列

        *

        * @return

        */

        @Bean

        public AmqpAdmin amqpAdmin() {

            return new RabbitAdmin(connectionFactory());

        }

        // 必须是prototype类型

        @Bean

        @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)

        public RabbitTemplate rabbitTemplate() {

            RabbitTemplate template = new RabbitTemplate(connectionFactory());

            return template;

        }

        /**

           * FanoutExchange: 将消息分发到所有的绑定队列，无routingkey的概念

            * HeadersExchange：通过添加属性key-value匹配

            * DirectExchange:按照routingkey分发到指定队列

            * TopicExchange:多关键字匹配

            * @return

            */

            @Bean

            public DirectExchange defaultExchange() {

                return new DirectExchange(MqMetaConstant.EXCHANGE_EMAIL);

            }

            @Bean

            public FanoutExchange adminMessageExchange() {

                return new FanoutExchange(MqMetaConstant.EXCHANGE_ADMIN_MESSAGE); }

         @Bean

        public Queue queue() {

            return new Queue(MqMetaConstant.QUEUE_EMAIL, true); // 队列持久 }

        @Bean

        public Queue adminMessageQueue() {

            return new Queue(MqMetaConstant.QUEUE_ADMIN_MESSAGE, true); // 队列持久 }

        @Bean

        public Binding binding() {

            return BindingBuilder.bind(queue()).to(defaultExchange()).with(MqMetaConstant.ROUTINGKEY_EMAIL); }

        @Bean

        public Binding adminMessageBinding() {

            return BindingBuilder.bind(adminMessageQueue()).to(adminMessageExchange()); }

 }

import java.util.List;

import java.util.UUID;

import com.wow.common.mq.vo.*;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

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 com.wow.common.util.CollectionUtil;

import com.wow.common.util.JsonUtil;

/**

* 发送消息到rabbitmq

* @author wangming@wowdsgn.com

* @date 2017年11月6日 下午5:41:58

*/

@Component

public class MqProducer implements RabbitTemplate.ConfirmCallback {

private static final Logger logger=LoggerFactory.getLogger(MqProducer.class);

private RabbitTemplate rabbitTemplate;

/** * 构造方法注入 */

@Autowired

public MqProducer(RabbitTemplate rabbitTemplate) {

    this.rabbitTemplate = rabbitTemplate;

    rabbitTemplate.setConfirmCallback(this);

   //rabbitTemplate如果为单例的话，那回调就是最后设置的内容      rabbitTemplate.setReturnCallback((message,replyCode,replyText,exchange,routingKey)->{ logger.info("没有找到任何匹配的队列！"+ "message:"+message+ ",replyCode:"+replyCode+ ",replyText:"+replyText+ ",exchange:"+exchange+ ",routingKey:"+routingKey); }); }

/**

* 发送邮件消息

*

* @param emailVO

*/

public void sendEmail(EmailVO emailVO) {

    if(emailVO==null) { logger.warn("email对象为空，拒绝发送邮件消息"); return ; }

    CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());

    String message=JsonUtil.pojo2Json(emailVO);

    logger.info("开始发送mq邮件消息：id:{}，内容：{}，",correlationId,message);    rabbitTemplate.convertAndSend(MqMetaConstant.EXCHANGE_EMAIL, MqMetaConstant.ROUTINGKEY_EMAIL,message , correlationId); }            

/**

* 发送后台消息

* @param messageList

*/

public void sendAdminMessage(AdminMessageVO messageVO) {

        CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());

        String message=JsonUtil.pojo2Json(messageVO);

        logger.info("开始发送后台消息：id:{}，内容：{}，",correlationId,message);         rabbitTemplate.convertAndSend(MqMetaConstant.EXCHANGE_ADMIN_MESSAGE, "",message , correlationId); }

      /** * 回调 */

    @Override

    public void confirm(CorrelationData correlationData, boolean ack, String cause) {

        if (ack) {

            logger.info("mq消息发送成功---确认");

        } else {

            logger.info("mq消息发送失败:id:{}，cause：{}" , correlationData,cause);

        }

    }

}