spring boot 整合 rabbit mq 详解

spring boot 整合rabbitmq 实战

1. pom文件

<!-- ================== rabbitMq====================-->
<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-amqp</artifactId>
	<version>2.0.0.RELEASE</version>
</dependency>

2. rabbitmq 配置文件

2.1. RabbitConfig总配置文件

package com.doyd.config.rabbitmq;

import com.doyd.config.rabbitmq.listener.CardCodeChannel;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * RabbitMQ 配置类
 */
@Configuration
@ConfigurationProperties(prefix = "spring.rabbitmq")
@Slf4j
@Getter
@Setter
public class RabbitConfig {
    /**主机*/
    private String host;
    /**端口*/
    private int port;
    /**用户名*/
    private String username;
    /**密码*/
    private String password;
    /**卡券空间*/
    private String virtualCard;

    /**
     * 功能描述: 卡券rabbitMq配置
     * @auther: huangsenming
     * @date: 2018/12/1 12:50
     */
    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        setCommon(connectionFactory);
        connectionFactory.setVirtualHost(virtualCard);
        connectionFactory.addChannelListener(new CardCodeChannel());
        return connectionFactory;
    }
    /**
     * 功能描述: 卡券rabbitMq配置
     * @auther: huangsenming
     * @date: 2018/12/1 12:50
     */
    @Bean
    public RabbitTemplate rabbitCardTemplate() {
        return new RabbitTemplate(connectionFactory());
    }
    /**
     * 功能描述:rabbit公共配置
     * @auther: huangsenming
     * @date: 2018/10/15 15:56
     */
    private void setCommon(CachingConnectionFactory connectionFactory) {
        connectionFactory.setHost(host);
        connectionFactory.setPort(port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
    }
}

2.2. properties配置文件

spring.rabbitmq.password=root
spring.rabbitmq.username=root
spring.rabbitmq.host= 127.0.0.1
spring.rabbitmq.port= 5672
spring.rabbitmq.virtualCard=card_code

3. 配置队列监听器

package com.doyd.queue;

import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

/**
 * @Auther: huangsenming
 * @Date: 2018/12/1 12:54
 * @Description:
 */
@Service
@Slf4j
public class CardQueueListen {
    @Autowired
    private RabbitTemplate rabbitCardTemplate;
    @RabbitListener(queues = "queue.card.off")
    public void cardOffDeal(Message message, Channel channel ){
        SimpleMessageConverter d=new SimpleMessageConverter();
        Object messObj=d.fromMessage(message);
        log.info("接收到的消息{}",JSON.toJSONString(messObj));
    }

}

4. 优化

如果只是这样配置需要单独配置路由和队列，增加一个channelListener就可以在连接建立的时候，对路由以及队列进行相关的配置

package com.doyd.config.rabbitmq.listener;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.ChannelListener;
import org.springframework.context.annotation.Configuration;

import java.io.IOException;

/**
 * @Auther: huangsenming
 * @Date: 2018/12/1 12:58
 * @Description: rabbitMq初始化路由以及队列数据
 */
@Slf4j
@Configuration
public class CardCodeChannelListener implements ChannelListener {
    @Override
    public void onCreate(Channel channel, boolean transactional) {
        try {
            log.info("卡券相关的路由队列绑定开始");
            cardOffBind(channel,transactional);
            log.info("卡券相关的路由队列绑定完成");

        } catch (Exception e) {
            log.error("路由队列绑定异常",e);
        }
    }

    private void  cardOffBind(Channel channel, boolean transactional) throws Exception{
        channel.exchangeDeclare("direct.card.off",BuiltinExchangeType.DIRECT);
        channel.queueDeclare("queue.card.off",true,true,false,null);
        channel.queueBind("queue.card.off","direct.card.off",  "card.off");
    }
}