可自动恢复的JMS消息收发类，完美处理异常

Question

呵呵又是一个“完美”，老王卖瓜，自卖自夸。搞了两天搞出来的，经过诸多断连测试，暂时没有发现问题。简介：封装了JMS主题发布/订阅方式，消息格式规定为ObjectMessage，JMS服务器采用的是OpenMQ，发送方和接收方都能够自动恢复重连，缺点是断线期间所有消息自动丢弃（不过在我参与的项目里，无所谓啦）。废话少说，贴代码，希望对各位有用，另外代码肯定还是有尚未发现的问题的，欢迎各位指导。
import java.io.Serializable;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import org.apache.log4j.Logger;/**
 * 消息发送器，能够断连自动恢复，不过断连情况下的待发送消息自动丢弃
 * @author
 */
public class MessageSender {    private static Logger logger = Logger.getLogger(MessageSender.class);
    private Connection connection = null;
    private ConnectionFactory connectionFactory = null;
    private Destination destination = null;
    /**
     * 是否已连接标识
     */
    private boolean connected = false;
    /**
     * JMS服务器地址，IP:Port格式
     */
    private String serverAddr;
    /**
     * 主题名
     */
    private String topicName;    public MessageSender(String serverAddr, String topicName) {
        this.serverAddr = serverAddr;
        this.topicName = topicName;
    }    /**
     * 初始化
     */
    private void initialize() {
        try {
            // 初始化连接工厂
            com.sun.messaging.ConnectionFactory factory = new com.sun.messaging.ConnectionFactory();
            factory.setProperty(com.sun.messaging.ConnectionConfiguration.imqAddressList, serverAddr);
            factory.setProperty(com.sun.messaging.ConnectionConfiguration.imqReconnectEnabled, "true");
            connectionFactory = factory;            // 初始化连接
            connection = connectionFactory.createConnection();            // 初始化目的地
            destination = new com.sun.messaging.Topic(topicName);            connected = true;
        } catch (JMSException ex) {
            logger.error("初始化失败：" + ex);
            cleanup();
        }
    }    /**
     * 清理
     */
    public void cleanup() {
        if (connection != null) {
            try {
                connection.close();
            } catch (JMSException ex) {
                logger.error("连接关闭失败：" + ex);
            }
        }
        connected = false;
    }    public void send(Serializable object) {
        if (!connected) {
            initialize();
        }        if (connected) {
            Session session = null;
            MessageProducer producer = null;
            try {
                session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                producer = session.createProducer(destination);
                ObjectMessage message = session.createObjectMessage(object);
                producer.send(message);
                logger.info("已发送：" + object);
            } catch (JMSException ex) {
                logger.error("发送失败：" + ex);
                cleanup();
            } finally {
                if (producer != null) {
                    try {
                        producer.close();
                    } catch (JMSException ex) {
                        logger.error("生产者关闭失败：" + ex);
                    }
                }
                if (session != null) {
                    try {
                        session.close();
                    } catch (JMSException ex) {
                        logger.error("会话关闭失败：" + ex);
                    }
                }
            }
        } else {
            logger.error("初始化失败导致发送失败");
        }
    }
}import java.io.Serializable;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
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.ObjectMessage;
import javax.jms.Session;
import org.apache.log4j.Logger;/**
 * 消息接收器，能够断连自动恢复
 * @author
 */
public abstract class MessageReceiver implements MessageListener, ExceptionListener {    private static Logger logger = Logger.getLogger(MessageReceiver.class);
    private Connection connection = null;
    private ConnectionFactory connectionFactory = null;
    private Destination destination = null;
    private Session session = null;
    private MessageConsumer consumer = null;
    /**
     * 是否已连接标识
     */
    private boolean connected = false;
    /**
     * 是否需要停止
     */
    private boolean needStop = false;
    /**
     * JMS服务器地址，IP:Port格式
     */
    private String serverAddr;
    /**
     * 主题名
     */
    private String topicName;    public MessageReceiver(String serverAddr, String topicName) {
        this.serverAddr = serverAddr;
        this.topicName = topicName;
    }    /**
     * 启动
     */
    public void start() {
        needStop = false;
        (new Thread(new Runnable() {            public void run() {
                initialize();
            }
        })).start();
    }    /**
     * 停止
     */
    public void stop() {
        needStop = true;
        cleanup();
    }    /**
     * 初始化，注意第一次初始化应该放在单独的线程中执行，避免阻塞
     */
    private void initialize() {
        while ((!connected) && (!needStop)) {
            try {
                // 初始化连接工厂
                com.sun.messaging.ConnectionFactory factory = new com.sun.messaging.ConnectionFactory();
                factory.setProperty(com.sun.messaging.ConnectionConfiguration.imqAddressList, serverAddr);
                factory.setProperty(com.sun.messaging.ConnectionConfiguration.imqReconnectEnabled, "true");
                connectionFactory = factory;                // 初始化连接
                connection = connectionFactory.createConnection();                connection.setExceptionListener(this);                // 初始化目的地
                destination = new com.sun.messaging.Topic(topicName);                // 注册消息监听者
                session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                consumer = session.createConsumer(destination);
                consumer.setMessageListener(this);                // 开始接收
                connection.start();                connected = true;
            } catch (JMSException ex) {
                logger.error("初始化失败：" + ex);
                cleanup();
            }
            try {
                Thread.sleep(1000);
            } catch (InterruptedException ex) {
                logger.error("睡眠被打断：" + ex);
            }
        }
    }    /**
     * 清除
     */
    public void cleanup() {
        if (consumer != null) {
            try {
                consumer.close();
            } catch (JMSException ex) {
                logger.error("生产者关闭失败：" + ex);
            }
        }
        if (session != null) {
            try {
                session.close();
            } catch (JMSException ex) {
                logger.error("会话关闭失败：" + ex);
            }
        }
        if (connection != null) {
            try {
                connection.close();
            } catch (JMSException ex) {
                logger.error("连接关闭失败：" + ex);
            }
        }
        connected = false;
    }    public void onException(JMSException exception) {
        logger.error("异常监听器接收到异常：" + exception);
        cleanup();
        initialize();
    }    public void onMessage(Message message) {
        try {
            Serializable object = ((ObjectMessage) message).getObject();
            logger.info("已接收：" + object);
            processMessage(object);
        } catch (JMSException ex) {
            logger.error("消息接收异常：" + ex);
        }
    }    public abstract void processMessage(Serializable object);
}
使用方法：发送消息方：
// 创建消息发送器，sender实例是线程安全的（共享连接）
MessageSender sender = new MessageSender(JMS服务器地址, 主题名);// 发送消息，该方法是非阻塞的，发送失败的消息自动丢弃
sender.send(消息);// 清除工作，注意如果不用了一定要调用以便释放连接
sender.cleanup();
接受消息方：
// 创建消息接收器并重载processMessage方法
receiver = new MessageReceiver(JMS服务器地址, 主题名) { @Override
public void processMessage(final Serializable object) {
// 此处处理受到的消息
}
};// 启动消息接收器，注意该方法是非阻塞的
receiver.start();// 停止消息接收器
receiver.stop();

Answer 1

呵呵又是一个“完美”，老王卖瓜，自卖自夸。搞了两天搞出来的，经过诸多断连测试，暂时没有发现问题。简介：封装了JMS主题发布/订阅方式，消息格式规定为ObjectMessage，JMS服务器采用的是OpenMQ，发送方和接收方都能够自动恢复重连，缺点是断线期间所有消息自动丢弃（不过在我参与的项目里，无所谓啦）。废话少说，贴代码，希望对各位有用，另外代码肯定还是有尚未发现的问题的，欢迎各位指导。
import java.io.Serializable;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import org.apache.log4j.Logger;/**
 * 消息发送器，能够断连自动恢复，不过断连情况下的待发送消息自动丢弃
 * @author
 */
public class MessageSender {    private static Logger logger = Logger.getLogger(MessageSender.class);
    private Connection connection = null;
    private ConnectionFactory connectionFactory = null;
    private Destination destination = null;
    /**
     * 是否已连接标识
     */
    private boolean connected = false;
    /**
     * JMS服务器地址，IP:Port格式
     */
    private String serverAddr;
    /**
     * 主题名
     */
    private String topicName;    public MessageSender(String serverAddr, String topicName) {
        this.serverAddr = serverAddr;
        this.topicName = topicName;
    }    /**
     * 初始化
     */
    private void initialize() {
        try {
            // 初始化连接工厂
            com.sun.messaging.ConnectionFactory factory = new com.sun.messaging.ConnectionFactory();
            factory.setProperty(com.sun.messaging.ConnectionConfiguration.imqAddressList, serverAddr);
            factory.setProperty(com.sun.messaging.ConnectionConfiguration.imqReconnectEnabled, "true");
            connectionFactory = factory;            // 初始化连接
            connection = connectionFactory.createConnection();            // 初始化目的地
            destination = new com.sun.messaging.Topic(topicName);            connected = true;
        } catch (JMSException ex) {
            logger.error("初始化失败：" + ex);
            cleanup();
        }
    }    /**
     * 清理
     */
    public void cleanup() {
        if (connection != null) {
            try {
                connection.close();
            } catch (JMSException ex) {
                logger.error("连接关闭失败：" + ex);
            }
        }
        connected = false;
    }    public void send(Serializable object) {
        if (!connected) {
            initialize();
        }        if (connected) {
            Session session = null;
            MessageProducer producer = null;
            try {
                session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                producer = session.createProducer(destination);
                ObjectMessage message = session.createObjectMessage(object);
                producer.send(message);
                logger.info("已发送：" + object);
            } catch (JMSException ex) {
                logger.error("发送失败：" + ex);
                cleanup();
            } finally {
                if (producer != null) {
                    try {
                        producer.close();
                    } catch (JMSException ex) {
                        logger.error("生产者关闭失败：" + ex);
                    }
                }
                if (session != null) {
                    try {
                        session.close();
                    } catch (JMSException ex) {
                        logger.error("会话关闭失败：" + ex);
                    }
                }
            }
        } else {
            logger.error("初始化失败导致发送失败");
        }
    }
}import java.io.Serializable;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
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.ObjectMessage;
import javax.jms.Session;
import org.apache.log4j.Logger;/**
 * 消息接收器，能够断连自动恢复
 * @author
 */
public abstract class MessageReceiver implements MessageListener, ExceptionListener {    private static Logger logger = Logger.getLogger(MessageReceiver.class);
    private Connection connection = null;
    private ConnectionFactory connectionFactory = null;
    private Destination destination = null;
    private Session session = null;
    private MessageConsumer consumer = null;
    /**
     * 是否已连接标识
     */
    private boolean connected = false;
    /**
     * 是否需要停止
     */
    private boolean needStop = false;
    /**
     * JMS服务器地址，IP:Port格式
     */
    private String serverAddr;
    /**
     * 主题名
     */
    private String topicName;    public MessageReceiver(String serverAddr, String topicName) {
        this.serverAddr = serverAddr;
        this.topicName = topicName;
    }    /**
     * 启动
     */
    public void start() {
        needStop = false;
        (new Thread(new Runnable() {            public void run() {
                initialize();
            }
        })).start();
    }    /**
     * 停止
     */
    public void stop() {
        needStop = true;
        cleanup();
    }    /**
     * 初始化，注意第一次初始化应该放在单独的线程中执行，避免阻塞
     */
    private void initialize() {
        while ((!connected) && (!needStop)) {
            try {
                // 初始化连接工厂
                com.sun.messaging.ConnectionFactory factory = new com.sun.messaging.ConnectionFactory();
                factory.setProperty(com.sun.messaging.ConnectionConfiguration.imqAddressList, serverAddr);
                factory.setProperty(com.sun.messaging.ConnectionConfiguration.imqReconnectEnabled, "true");
                connectionFactory = factory;                // 初始化连接
                connection = connectionFactory.createConnection();                connection.setExceptionListener(this);                // 初始化目的地
                destination = new com.sun.messaging.Topic(topicName);                // 注册消息监听者
                session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                consumer = session.createConsumer(destination);
                consumer.setMessageListener(this);                // 开始接收
                connection.start();                connected = true;
            } catch (JMSException ex) {
                logger.error("初始化失败：" + ex);
                cleanup();
            }
            try {
                Thread.sleep(1000);
            } catch (InterruptedException ex) {
                logger.error("睡眠被打断：" + ex);
            }
        }
    }    /**
     * 清除
     */
    public void cleanup() {
        if (consumer != null) {
            try {
                consumer.close();
            } catch (JMSException ex) {
                logger.error("生产者关闭失败：" + ex);
            }
        }
        if (session != null) {
            try {
                session.close();
            } catch (JMSException ex) {
                logger.error("会话关闭失败：" + ex);
            }
        }
        if (connection != null) {
            try {
                connection.close();
            } catch (JMSException ex) {
                logger.error("连接关闭失败：" + ex);
            }
        }
        connected = false;
    }    public void onException(JMSException exception) {
        logger.error("异常监听器接收到异常：" + exception);
        cleanup();
        initialize();
    }    public void onMessage(Message message) {
        try {
            Serializable object = ((ObjectMessage) message).getObject();
            logger.info("已接收：" + object);
            processMessage(object);
        } catch (JMSException ex) {
            logger.error("消息接收异常：" + ex);
        }
    }    public abstract void processMessage(Serializable object);
}
使用方法：发送消息方：
// 创建消息发送器，sender实例是线程安全的（共享连接）
MessageSender sender = new MessageSender(JMS服务器地址, 主题名);// 发送消息，该方法是非阻塞的，发送失败的消息自动丢弃
sender.send(消息);// 清除工作，注意如果不用了一定要调用以便释放连接
sender.cleanup();
接受消息方：
// 创建消息接收器并重载processMessage方法
receiver = new MessageReceiver(JMS服务器地址, 主题名) { @Override
public void processMessage(final Serializable object) {
// 此处处理受到的消息
}
};// 启动消息接收器，注意该方法是非阻塞的
receiver.start();// 停止消息接收器
receiver.stop();

Answer 2

以前用weblogic的JMS,订阅发布的消息断线后会自动重发的