动态管理RabbitMQ队列监听

怼烎@ 2021-12-17 12:19 412阅读 0赞

需求背景：最近项目要求，动态监听MQ，实现程序新增删除MQ队列的时候，可以动态调整监听。

废话不说了，贴代码：能力有限，欢迎批判。

import java.io.IOException;
    import java.util.ArrayList;
    import java.util.Arrays;
    import java.util.List;
    import java.util.Map;
    import java.util.Set;
    import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.TimeoutException;
    
    import org.apache.commons.lang3.StringUtils;
    import org.apache.log4j.Logger;
    import org.springframework.amqp.AmqpException;
    import org.springframework.amqp.core.AcknowledgeMode;
    import org.springframework.amqp.core.MessageListener;
    import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
    import org.springframework.amqp.rabbit.connection.Connection;
    import org.springframework.amqp.rabbit.connection.ConnectionFactory;
    import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
    import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
    
    class  
    {
        /*initListenerContainer and start listeners
         */
        private void initListenerContainer(){
            logger.info("---------------->>>>>>>>>>> 初始化RabbitMQ监听容器 <<<<<<<<<<<----------------");
            try {
                RabbitListenerEndpointRegistry rabbitListenerEndpointRegistry = DynamicListenerCfg.getRabbitListenerEndpointRegistry();//这里是一个单例，返回一个RabbitListenerEndpointRegistry 实例。
                           
                rabbitListenerEndpointRegistry.stop();
    
                SimpleMessageListenerContainer container = null;//监听容器
                //connection info
                QueueInfo info = new QueueInfo();
                info.setUserName("userName");
                info.setPassword("helloworld");
                info.setServerIp("192.168.0.110");
                info.setServerPort(5672);
                info.setVirtualHost("/");
                String[] queueNames = null;
                    
                queueNames = new String[] {"queueA","queueB"};
    
                if (queueNames != null && queueNames.length > 0) {
                    container = createListenerContainer(MessageListener, info, queueNames);
                    Assert.state(!container.isRunning(), String.format("监听容器%s正在运行！", container));
                    container.start();
                    listenerContainerList.add(container);
                }
    
                rabbitListenerEndpointRegistry.start();
        
            } catch (Exception e) {
                e.getSuppressed();
            }
            
        }
    
        //create container by queueName and lisenerInstance
        public SimpleMessageListenerContainer createListenerContainer(MessageListener listenerInstant,QueueInfo info, String ...queueNames) {
            SimpleMessageListenerContainer container = null;
            try {
                ConnectionFactory factory = connectionFactory(info.getServerPort(),info.getUserName(),info.getPassword(),info.getServerIp(),info.getVirtualHost());
                container = new SimpleMessageListenerContainer(factory);
                container.addQueueNames(queueNames);
                container.setMessageListener(listenerInstant);//配置自定义的监听类
                container.setAutoStartup(isAutoStartUp);
            } catch (Exception e) {
                e.printStackTrace();
            }
            return container;
            
        }
    
        //create queue and bind listener
    	public boolean addListenerQueue(QueueInfo info) throws AmqpException, IOException {
    		boolean rst = false;
    		//create queue
    		
    		Connection connection = null;
    		try {
    			connection = connectionFactory(info.getServerPort(),info.getUserName(),info.getPassword(),info.getServerIp(),info.getVirtualHost()).createConnection();
                //如果删除队列，这里改成调用queueDelete方法即可（先判断下messageCount是否还有消息未处理）
    			connection.createChannel(false).queueDeclare(info.getQueueName(), false, false, true, null);
    
    			//监听队列	
    		        listenerContainerList.forEach(a->a.stop());
                //删除的时候这里改为调用removeQueueNames即可.
    			listenerContainerList.forEach(a->a.addQueueNames(info.getQueueName()));
    			listenerContainerList.forEach(a->a.start());
    
    			rst = true;
    		} catch (Exception e) {
    			rst = false;
    			e.printStackTrace();
    		} finally {
    			if (connection != null) {
    				connection.close();
    			}
    		}
    		return rst;
    	}
    }