RabbitMQ之（3）接收端

逃离我推掉我的手 2021-09-28 03:50 179阅读 0赞

package rabbitmq;
    
    import org.apache.log4j.LogManager;
    import org.apache.log4j.Logger;
    
    import com.rabbitmq.client.ConsumerCancelledException;
    import com.rabbitmq.client.QueueingConsumer;
    import com.rabbitmq.client.ShutdownSignalException;
    /**
     * @ClassName: RabbitMQReceive 
     * @Description: RabbitMQ接收端
     * @author wxy 
     * @date 2019年1月24日 下午1:30:18 
     */
    public class RabbitMQReceive extends RabbitMQConnection implements Runnable{
    	protected final Logger LOG = LogManager.getLogger(RabbitMQReceive.class);
    	//消费者
    	private QueueingConsumer consumer;
    	//消息监听者
    	private RabbitMQListener listener;
    	
    	public RabbitMQReceive(RabbitMQListener listener) {
    		super();
    		this.listener = listener;
    		this.init();
    	}
    	
    	public RabbitMQReceive(String rabbitmq_url, int rabbitmq_port, String userName, String password,
    			String rabbit_queue_name, RabbitMQListener listener) {
    		super(rabbitmq_url, rabbitmq_port, userName, password, rabbit_queue_name);
    		this.listener = listener;
    		this.init();
    	}
    	
    	public void init() {
    		super.initConnection();
    		getReceive();
    	}
    	
    	public void getReceive() {
    		LOG.info("启动接收端线程");
    		new Thread(this).start();
    	}
    	
    	@Override
    	public void run() {
    		while(true) {
    			if(null != listener) {
    				if(null == consumer) {
    					try {
    						consumer = new QueueingConsumer(channel);
    						channel.basicConsume(rabbit_queue_name, true, consumer);
    					} catch (Exception e) {
    						LOG.error(e);
    						if(null != consumer) {
    							consumer = null;
    						}
    					}
    				}
    				
    				try {
    					QueueingConsumer.Delivery delivery = consumer.nextDelivery();
    					if(null == listener) {
    						listener = new TestListener();
    					}
    					listener.receiveMessage(new String(delivery.getBody()));
    				} catch (ShutdownSignalException e) {
    					e.printStackTrace();
    				} catch (ConsumerCancelledException e) {
    					e.printStackTrace();
    				} catch (InterruptedException e) {
    					e.printStackTrace();
    				} catch(Exception e) {
    					LOG.error(e);
    					this.close();
    					if(null != channel) {
    						channel = null;
    					}
    					getChannel();
    				}
    			}
    			
    			try {
    				Thread.sleep(1000);
    			} catch (Exception e) {
    				e.printStackTrace();
    			}
    		}
    	}
    	
    	private class TestListener implements RabbitMQListener {
    		
    		public TestListener(){
    		}
    		
    		@Override
    		public void receiveMessage(String message) {
    			LOG.info("接收消息：" + message);
    		}
    		
    	}
    	
    	public static void main(String[] args) {
    		RabbitMQReceive receive = new RabbitMQReceive(null);
    		receive.init();
    		receive.getReceive();
    	}
    }