关于java:异步消息处理机制之activeMQ应用实例

上篇说了 KAFKA 利用实例，本篇承接上篇，着重形容 activeMQ 音讯机制的利用。
KAFKA 和 MQ 同为数据异步解决中间件，实质都是对音讯的异步解决，异步通信、削谷填峰，高并发状况下的数据处理机制。他们的不同之处在于解决数据量的大小。
MQ 和 KAFKA 相比拟，KAFKA 解决的数据量更大.
下图为 activeMQ 利用目录：

ActiveMQ 客户端，对连贯和会话的治理

/**
 * 
 */
package com.ustcinfo.kanms.alarmcollector.activemq;

import javax.jms.Connection;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.log4j.Logger;

import com.ustcinfo.kanms.alarmcollector.main.GlobleConfiguration;

/**
 * ================================================= <br>
 * 工程：GessAlarmCollector <br>
 * 类名：ActiveMQClient <br>
 * 作者：dlzhang <br>
 * 工夫：2014-8-28 下午 05:37:21<br>
 * 版本：Version 1.0 <br>
<br>
 * 形容：ActiveMQ 客户端，对连贯和会话的治理，保障全局只有一个会话被创立，缩小服务端压力、节俭资源 <br>
 * ================================================= <br>
 */
public class ActiveMQClient{private static final Logger logger = Logger.getLogger(ActiveMQClient.class);
    
    
    private String url;
    private String user;
    private String passwd;
    
    private ActiveMQConnectionFactory connFactory;
    private Connection conn;
    private Session session;
    
    private boolean isConn;
    
    public ActiveMQClient(){
        // 初始化参数
        this.url = GlobleConfiguration.getInstance().getActiveMqUrl();
        this.user = GlobleConfiguration.getInstance().getActiveMqUser();
        this.passwd = GlobleConfiguration.getInstance().getActiveMqPasswd();
    }
    
    
    /**
     * 建设连贯
     */
    protected synchronized void buildConnect() {if(isConn)
            return;
        try {logger.debug("建设连贯，user=" + user + ", passwd=" + passwd + ", url=" + url);
            connFactory = new ActiveMQConnectionFactory(user, passwd, url);
            conn = connFactory.createConnection();
            conn.start();
            session = conn.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
            isConn = true;
            logger.info("建设连贯胜利");
        } catch (JMSException e) {logger.error("建设连贯失败：" + e.getMessage(), e);
            isConn = false;
        }
    }
    
    /**
     * 敞开连贯
     */
    public synchronized void close() {
        try {if(null != session)
                session.close();
            if(null != conn)
                conn.close();} catch (JMSException e) {logger.error("敞开连贯失败：" + e.getMessage(), e);
        } finally {
            session = null;
            conn = null;
            connFactory = null;
            isConn = false;
        }
    }

    /**
     * @return the url
     */
    public String getUrl() {return url;}

    /**
     * @param url the url to set
     */
    public void setUrl(String url) {this.url = url;}

    /**
     * @return the user
     */
    public String getUser() {return user;}

    /**
     * @param user the user to set
     */
    public void setUser(String user) {this.user = user;}

    /**
     * @return the passwd
     */
    public String getPasswd() {return passwd;}

    /**
     * @param passwd the passwd to set
     */
    public void setPasswd(String passwd) {this.passwd = passwd;}

    /**
     * @return the connFactory
     */
    public ActiveMQConnectionFactory getConnFactory() {if(!isConn)
            buildConnect();
        if(null == connFactory && isConn) {this.close();
            this.buildConnect();}
        return connFactory;
    }

    /**
     * @param connFactory the connFactory to set
     */
    public void setConnFactory(ActiveMQConnectionFactory connFactory) {this.connFactory = connFactory;}

    /**
     * @return the conn
     */
    public Connection getConn() {if(!isConn)
            buildConnect();
        if(null == conn && isConn) {this.close();
            this.buildConnect();}
        return conn;
    }

    /**
     * @param conn the conn to set
     */
    public void setConn(Connection conn) {this.conn = conn;}

    /**
     * @return the session
     */
    public Session getSession() {if(!isConn)
            buildConnect();
        if(null == session && isConn) {this.close();
            this.buildConnect();}
        return session;
    }

    /**
     * @param session the session to set
     */
    public void setSession(Session session) {this.session = session;}

    /**
     * @return the isOpen
     */
    public boolean isConn() {return isConn;}

    /**
     * @param isOpen the isOpen to set
     */
    public void setConn(boolean isConn) {this.isConn = isConn;}


}

activemq 音讯主体的定义和申明

/**
 * 
 */
package com.ustcinfo.kanms.alarmcollector.activemq;

import java.io.Serializable;

import javax.jms.BytesMessage;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.log4j.Logger;

import com.ustcinfo.kanms.alarmcollector.main.GlobleConfiguration;

/**
 * ================================================= <br>
 * 工程：GessAlarmCollector <br>
 * 类名：ActiveMQSender <br>
 * 作者：dlzhang <br>
 * 工夫：2014-8-29 下午 01:02:40<br>
 * 版本：Version 1.0 <br>
 <br>
 * 形容：形容该文件的作用 <br>
 * ================================================= <br>
 */
public class ActiveMQSender {private static final Logger logger = Logger.getLogger(ActiveMQSender.class);
    private ActiveMQClient mqClient;
    private Session session;
    private String sendQueueName;
    private Destination dest;
    private MessageProducer producer;
    
    public ActiveMQSender() {
        // 初始化参数
        this.mqClient = new ActiveMQClient();
        this.session = this.mqClient.getSession();
        this.sendQueueName = GlobleConfiguration.getInstance().getActiveMqSendQueueName();
        try {dest = session.createQueue(sendQueueName);
            producer = session.createProducer(dest);
        } catch (JMSException e) {logger.error(e.getMessage(), e);
        }
    }

//    /**
//     * 测试函数
//     * @param args
//     */
//    public static void main(String[] args) {
//        // 初始化零碎配置文件
//        GlobleConfiguration.getInstance().initSysConfig();
//        
//        ActiveMQSender sender = new ActiveMQSender();
////        for(int i=0; i<100; i++) {////            sender.sendTextMessage("这是一个测试");
//            sender.sendTextMessage("quit");
//            logger.debug("第" + ++num + "条音讯发送胜利");
////        }
//    }
//    private static long num;
    
    /**
     * @param msg
     */
    public void sendMessage(Message msg) {
        try {producer.send(msg);
        } catch (JMSException e) {logger.error(e.getMessage(), e);
        }
    }
    
    /**
     * @param text
     */
    public void sendTextMessage(String text) {
        TextMessage tMsg = null;
        try {tMsg = session.createTextMessage(text);
        } catch (JMSException e) {logger.error(e.getMessage(), e);
        }
        this.sendMessage(tMsg);
    }
    
    /**
     * @param bytes
     */
    public void sendbytesMessage(byte[] bytes) {
        BytesMessage bMsg = null;
        try {bMsg = session.createBytesMessage();
            bMsg.writeBytes(bytes);
        } catch (JMSException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();}
        this.sendMessage(bMsg);
    }
    
    /**
     * @param s
     */
    public void sendObjectMessage(Serializable s) {
        ObjectMessage oMsg = null;
        try {oMsg = session.createObjectMessage(s);
        } catch (JMSException e) {logger.error(e.getMessage(), e);
        }
        this.sendMessage(oMsg);
    }

    /**
     * @return the mqClient
     */
    public ActiveMQClient getMqClient() {return mqClient;}

    /**
     * @param mqClient the mqClient to set
     */
    public void setMqClient(ActiveMQClient mqClient) {this.mqClient = mqClient;}

    /**
     * @return the session
     */
    public Session getSession() {return session;}

    /**
     * @param session the session to set
     */
    public void setSession(Session session) {this.session = session;}

    /**
     * @return the sendQueueName
     */
    public String getSendQueueName() {return sendQueueName;}

    /**
     * @param sendQueueName the sendQueueName to set
     */
    public void setSendQueueName(String sendQueueName) {this.sendQueueName = sendQueueName;}

    /**
     * @return the dest
     */
    public Destination getDest() {return dest;}

    /**
     * @param dest the dest to set
     */
    public void setDest(Destination dest) {this.dest = dest;}

    /**
     * @return the producer
     */
    public MessageProducer getProducer() {return producer;}

    /**
     * @param producer the producer to set
     */
    public void setProducer(MessageProducer producer) {this.producer = producer;}
    
}

音讯接受器，应用监听的形式承受音讯

/**
 * 
 */
package com.ustcinfo.kanms.alarmcollector.activemq;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.log4j.Logger;

import com.ustcinfo.kanms.alarmcollector.framework.BaseThread;
import com.ustcinfo.kanms.alarmcollector.framework.GesAlarmContainer;
import com.ustcinfo.kanms.alarmcollector.framework.ReadXML;
import com.ustcinfo.kanms.alarmcollector.kafka.ProducerHandler;
import com.ustcinfo.kanms.alarmcollector.main.GlobleConfiguration;
import com.ustcinfo.kanms.alarmcollector.model.GesAlarm;
import com.ustcinfo.kanms.alarmcollector.util.FileWriteThread;


/**
 * ================================================= <br>
 * 工程：AlarmAutoReceiver <br>
 * 类名：ActiveMQReceiver <br>
 * 作者：lt <br>
 * 工夫：2019-9-23 下午 08:49:38<br>
 * 版本：Version 1.0 <br>
<br>
 * 形容：告警音讯接受器，应用监听的形式承受音讯 <br>
 * 实现 ExceptionListener 接口，对服务器进行监听，实现主动重连性能 <br>
 * ================================================= <br>
 */
public class AlarmAutoReceiverThread{private static final Logger logger = Logger.getLogger(AlarmAutoReceiverThread.class);
     private final FileWriteThread filewrite = new FileWriteThread();// 建设一个新的线程写文件
     public AlarmAutoReceiverThread(int n) {for(int i=0;i<n;i++){new AlarmAutoReceiver().start();
             logger.info(">>>>> 胜利启动监听线程: 线程"+i);
         }
     }
     class AlarmAutoReceiver extends BaseThread implements MessageListener, ExceptionListener {private  final Logger log = Logger.getLogger(AlarmAutoReceiver.class);
        private ActiveMQClient mqClient;
        private GesAlarmContainer gesAlarmContainer;
        private Session session;
        private String recvQueueName;
        private Destination dest;
        private MessageConsumer consumer;
        
        public AlarmAutoReceiver() {
            // 初始化参数
            this.mqClient = new ActiveMQClient();
            this.gesAlarmContainer = GesAlarmContainer.getInstance();
            try {mqClient.getConn().setExceptionListener(this); // 设置监听
            } catch (JMSException e) {log.error(e.getMessage(), e);
            }
            this.session = this.mqClient.getSession();
            this.recvQueueName = GlobleConfiguration.getInstance().getAlarmRecvQueueName();
        }
        
        public AlarmAutoReceiver(ActiveMQClient mqClient) {
            // 初始化参数
            this.mqClient = mqClient;
            this.gesAlarmContainer = GesAlarmContainer.getInstance();
            try {mqClient.getConn().setExceptionListener(this); // 设置监听
            } catch (JMSException e) {log.error(e.getMessage(), e);
            }
            this.session = this.mqClient.getSession();
            this.recvQueueName = GlobleConfiguration.getInstance().getAlarmRecvQueueName();
        }
        
        /* (non-Javadoc)
         * @see java.lang.Thread#run()
         */
        @Override
        public void run() {log.debug("启动线程, Thread.currentThread().getName()=" + Thread.currentThread().getName());
            try {dest = session.createQueue(recvQueueName);
                consumer = session.createConsumer(dest);
                consumer.setMessageListener(this);
                // 阻塞线程，直到收到音讯为“quit”时被唤醒
                synchronized (this) {wait();
                }
                log.debug("完结线程, Thread.currentThread().getName()=" + Thread.currentThread().getName());
                mqClient.close();} catch (JMSException e) {log.error("创立消费者失败：" + e.getMessage(), e);
            } catch (InterruptedException e) {log.error("线程被中断", e);
            }
        }
        
        /* (non-Javadoc)
         * @see javax.jms.ExceptionListener#onException(javax.jms.JMSException)
         */
        @Override
        public void onException(JMSException e) {
            // 捕捉异样，连贯重连
            log.error("服务端异样", e);
            
            // 完结线程
            if(this.isAlive()) {synchronized (this) {this.notifyAll();
                }
            }
            
            // 每隔一分钟循环获取，直到获取连贯胜利
            long sleepTimeSec = 10;
            long sleepTimeMillis = sleepTimeSec * 1000;
            int reConnCnt = 0;
            while (true) {
                try {
                    reConnCnt++;
                    log.info("休眠" + sleepTimeSec + "秒 -_-~zZ");
                    Thread.sleep(sleepTimeMillis);
                    log.debug("开始从新获取连贯");
                    
                    // 先敞开连贯，再从新连贯
                    mqClient.close();
                    mqClient.buildConnect();
                    
                    if(mqClient.isConn()) {log.info("从新获取连贯胜利, 耗时：[" + reConnCnt * sleepTimeSec + "] 秒 ^_^");
                        // 从新创立监听
                        new AlarmAutoReceiver(mqClient).start();
                        break;
                    }
                    log.error("第" + reConnCnt + "次从新获取连贯失败 T_T");
                } catch (InterruptedException e1) {log.error(e1.getMessage(), e1);
                }
            }
        }
        
    //    /**
    //     * 测试函数
    //     * @param args
    //     */
    //    public static void main(String[] args) {
    //        // 初始化零碎配置文件
    //        GlobleConfiguration.getInstance().initSysConfig();
    //        
    //        ActiveMQReceiver recv = new ActiveMQReceiver();
    //        recv.start();
    //        logger.debug("消费者启动胜利");
    //    }
        
        /* (non-Javadoc)
         * @see javax.jms.MessageListener#onMessage(javax.jms.Message)
         */
        @Override
        public void onMessage(Message message) {log.info("-------------- 收到一条告警信息 -----------");
            if (message instanceof TextMessage){TextMessage  tm = (TextMessage)message;
                try {String xml = tm.getText();
                    if(xml!=null){
                        try{filewrite.handle(xml.substring(xml.indexOf("<ALARMS>"),xml.lastIndexOf("</ALARMS>")+9)+"\n--"+dateFormat(new Date()));
                        }catch(Exception e){log.error("告警日志写入出错!",e);
                            filewrite.handle(xml+"\n--"+dateFormat(new Date()));
                        }
                    }
                    // 当音讯为“quit”时，唤醒线程
                    if(xml != null && xml.equalsIgnoreCase("quit")) {logger.info(Thread.currentThread().getName() + "接管到的音讯为：" + xml + ", 开始退出线程");
                        synchronized (this) {notifyAll();
                        }
                        return;
                    }
                    /*String gesAlarmStr = ReadXML.obtainBodyInfo(xml.trim());
                    List<GesAlarm> gesAlarmList = ReadXML.getAlarmValue(gesAlarmStr);
                    if(gesAlarmList!=null&&gesAlarmList.size()>0){logger.debug("gesAlarmList.size()=" + gesAlarmList.size());
                        for(int i=0;i<gesAlarmList.size();i++){GesAlarm gesAlarm = new GesAlarm();
                            gesAlarm = gesAlarmList.get(i);
                            gesAlarmContainer.putGesAlarm(gesAlarm);
                        }            
                    }*/    
                    // 向 kafka 写入数据
                    new ProducerHandler("znwgAlarm",xml); 
                } catch (JMSException e) {} catch(Exception e){logger.error("承受告警信息出错！", e);
                }finally {}}
        }
        
        private  String dateFormat(Date date){SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            return sdf.format(date);
        }
    //    private long num;
        
        /**
         * @return the mqClient
         */
        public ActiveMQClient getMqClient() {return mqClient;}
    
        /**
         * @param mqClient the mqClient to set
         */
        public void setMqClient(ActiveMQClient mqClient) {this.mqClient = mqClient;}
    
        /**
         * @return the session
         */
        public Session getSession() {return session;}
    
        /**
         * @param session the session to set
         */
        public void setSession(Session session) {this.session = session;}
    
        /**
         * @return the recvQueueName
         */
        public String getRecvQueueName() {return recvQueueName;}
    
        /**
         * @param recvQueueName the recvQueueName to set
         */
        public void setRecvQueueName(String recvQueueName) {this.recvQueueName = recvQueueName;}
    
        /**
         * @return the dest
         */
        public Destination getDest() {return dest;}
    
        /**
         * @param dest the dest to set
         */
        public void setDest(Destination dest) {this.dest = dest;}
    
        /**
         * @return the consumer
         */
        public MessageConsumer getConsumer() {return consumer;}
    
        /**
         * @param consumer the consumer to set
         */
        public void setConsumer(MessageConsumer consumer) {this.consumer = consumer;}
        
        
    }
}