RabbitMQ連接池、生產者、消費者實例

1、本文分享RabbitMQ的工具類，經過實際項目長期測試，在此分享給發家，各位大神有什么建議請指正 ！！！

2、下面是鏈接池主要代碼：

import java.util.HashMap;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

/**
 * 獲取RabbitMq連接
 * @author skyfeng
 */
public class RabbitMqConnectFactory {
    static Logger log = LoggerFactory.getLogger(RabbitMqConnectFactory.class);
    /**
     * 緩存連接工廠,將建立的鏈接放入map緩存
     */
    private static Map<String, ConnectionFactory> connectionFactoryMap = new HashMap<String, ConnectionFactory>();
    /**
     * 根據rabbitMqName獲取一個連接，使用完記得要自己關閉連接 conn.close()
     */
    public static Connection getConnection(String rabbitMqName) {
        if(StringUtils.isEmpty(rabbitMqName)){
            log.error("rabbitMqName不能為空!");
            throw new java.lang.NullPointerException("rabbitMqName為空");
        }
        if(connectionFactoryMap.get(rabbitMqName)==null){
            initConnectionFactory(rabbitMqName);
        }
        ConnectionFactory connectionFactory = connectionFactoryMap.get(rabbitMqName);
        if(connectionFactory==null){
            log.info("沒有找到對應的rabbitmq,name={}",rabbitMqName);
        }
        try {
            return connectionFactory.newConnection();
        }catch (Exception e) {
            log.error("創建rabbitmq連接異常！",e);
            return null;
        }
    }
    /**
     * 初始化一個連接工廠
     * @param rabbitMqName
     */
    private static void initConnectionFactory(String rabbitMqName){
        
        try {
            ConnectionFactory factory = new ConnectionFactory();
            //新增代碼，如果連接斷開會自動重連
            //factory.setAutomaticRecoveryEnabled(true);
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            //factory.setVirtualHost(vhost);
            factory.setUsername("test");
            factory.setPassword("test");
            connectionFactoryMap.put(rabbitMqName, factory);
        } catch (Exception e) {
            e.printStackTrace();
        }finally{
        }
    }
    
}

 3、消費端的代碼：

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Consumer;

/**
 * RabbitMQq客戶端代碼
 * @author skyfeng
 *
 */
public class CustomerMqClient {

    final static Logger log = LoggerFactory.getLogger(CustomerMqClient.class);
    private final static String RABBITMQ_NAME = "mq_name";
    private final static String EXCHANGE_NAME = "Exchange_name";
    private final static String QUEUE_NAME = "queue_name";
    private static Channel channel = null;
    private static Connection connection = null;
    
    /**
     * 初始化客戶端代碼
     */
    public static void initClient() {
        //重新鏈接時判斷之前的channel是否關閉，沒有關閉先關閉
        if(null != channel  && channel.isOpen()){
            try {
                channel.close();
            } catch (Exception e) {
                log.error("mq name =[" +RABBITMQ_NAME+"] close old channel exception.e={}",e);
            }finally {
                channel = null;
            }
        }
        //重新鏈接時判斷之前的connection是否關閉，沒有關閉先關閉
        if (null != connection && connection.isOpen()) {
            try {
                connection.close();
            } catch (Exception e) {
                log.error("mq name =[" +RABBITMQ_NAME+"] close old connection exception.e={}",e);
            }finally{
                connection = null;
            }
        }
        //從鏈接池中獲取鏈接
        connection = RabbitMqConnectFactory.getConnection(RABBITMQ_NAME);
        try {
            channel = connection.createChannel();
            channel.exchangeDeclare(EXCHANGE_NAME, "topic", true);
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "#");//#號接收所有的數據
            Consumer consumer = new CustomerMqConsumer(channel);//具體的業務邏輯在CustomerMqConsumer中
            channel.basicConsume(QUEUE_NAME, false, consumer);
        } catch (Exception e) {
            log.error("CustomerMqClient mq client connection fail .....{}", e);
            //發生異常時，重連
            reConnect();
        }
    }

    // 異常時，重連的方法
    public static void reConnect() {
        log.error("等待5s后重連");
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
        }
        initClient();
    }

}

4、生產端代碼：

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.rabbitmq.client.AlreadyClosedException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

/**
 * 把數據發送到rabbitmq的exchange，
 */
public class SendToExchange {
    static Logger log = LoggerFactory.getLogger(SendToExchange.class);
    
    final static String TYPE = "topic";
    final static String CHARSET_UTF8 = "UTF-8";
    //MQ生產者exchange，把數據發給這個exchange
    final static String rabbitExchangeName = "ExchangeName";
    static boolean mqConnected = false;//mq當前處於連接狀態
    
    static Channel channel=null;
    static{
        init();
    }
    public static void init(){
        log.info(" rabbit mq init begin...");
        try {
            //在mq連接中斷后，發送程序判斷已經斷開，啟動重連的時候會執行
            if(channel!=null){
                try {
                    channel.close();
                } catch (Exception e) {
                    log.error("關閉老channel 異常",e);
                }finally{
                    channel = null;
                }
            }
            Connection connection = RabbitMqConnectFactory.getConnection("connection");
            channel = connection.createChannel();
            /*
             *這里只定義exchange，因為每個業務模塊都會從這里接入數據，所以不在這里定義隊列
             *隊列的定義在各個業務模塊自己的消費端定義
             */
            channel.exchangeDeclare(rabbitExchangeName, TYPE, true, false, null);
            log.info(" rabbit mq init OK");
            mqConnected = true;
        } catch (Exception e) {
            log.error("rabbitmq初始化錯誤",e);
            mqConnected = false;
        }
    }
    /**
     * 往rabbitmq發數據
     * @param message
     */
    public static void sendToRabbitMq(String message,String routingKey){
        try {
            if(StringUtils.isEmpty(message)){
                log.debug("message is empty");
                return;
            }
            channel.basicPublish(rabbitExchangeName, routingKey, null, message.getBytes(CHARSET_UTF8));
        }catch(AlreadyClosedException ex){
            log.error("往rabbitmq發數據報錯,可能連接已關閉,嘗試重連,data:",message,ex);
            init();
        }catch (Exception e) {
            log.error("往rabbitmq發數據報錯,data:",message,e);
        }
    }
}