Jms消费者模式

package com.ship;

import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Properties;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
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.activemq.ActiveMQConnectionFactory;
import org.springframework.stereotype.Controller;
import com.common.utils.PropertyLoader;

@Controller
public class JmsConsumer implements MessageListener {
    private java.sql.Connection con = null;
    private static JmsConsumer instance = null;

    public JmsConsumer() {
        if (instance != null)
            throw new RuntimeException();
        instance = this;
    }

    public JmsConsumer getInstance() {
        return instance;
    }

    @Override
    public void onMessage(Message message) {
        if (message instanceof TextMessage) {
            TextMessage txtMsg = (TextMessage) message;
            try {
                String content = txtMsg.getText();
                System.out.println("数据是：\n" + content);
                String[] arrcontent = content.split("\n");
                // con.setAutoCommit(false);
                // Statement statement = con.createStatement();
                for (int i = 0; i < arrcontent.length; i++) {
                    String[] shipinfo = arrcontent[i].split(",");
                    // System.out.println(shipinfo[0]); //测试船名
                    // 存储过程
                    // String sql = "call ganzhi()";
                    // 添加批处理
                    // statement.addBatch(sql);
                }
                // 批处理 因为存在读写延迟
                // statement.executeBatch();
                // con.commit();
                // statement.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    @PostConstruct
    public void init() {
        try {
            // 数据库连接
            buildSqlCon();
            // jms处理
            buildJmsCon();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void buildSqlCon() {
        // TODO Auto-generated method stub
        Properties jdbcprop = PropertyLoader.getPropertiesFromClassPath(
                "jdbc.properties", "UTF-8");
        String driverClassName = jdbcprop.getProperty("jdbc.driverClassName");
        String jdbcurl = jdbcprop.getProperty("jdbc.url");
        String uname = jdbcprop.getProperty("jdbc.username");
        String pwd = jdbcprop.getProperty("jdbc.password");
        try {
            Class.forName(driverClassName);
            con = DriverManager.getConnection(jdbcurl + "&user=" + uname
                    + "&password=" + pwd);
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }

    }

    public void buildJmsCon() {
        // 消费者的主要流程
        Connection connection = null;
        Properties properties = PropertyLoader.getPropertiesFromClassPath(
                "activemq.properties", "UTF-8");
        String topic = properties.getProperty("topic");
        String username = properties.getProperty("username");
        String password = properties.getProperty("password");
        String ipaddress = properties.getProperty("ipaddress");
        String port = properties.getProperty("port");
        String brokerURL = "failover://tcp://" + ipaddress + ":" + port;
        try {
            // 1.初始化connection工厂
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                    username, password, brokerURL);

            // 2.创建Connection
            connection = connectionFactory.createConnection();

            // 3.打开连接
            connection.start();

            System.out.println("连接成功..................");

            // 4.创建session
            Session session = connection.createSession(false,
                    Session.AUTO_ACKNOWLEDGE);

            // 5.创建消息目标
            Destination destination = session.createTopic(topic);

            // 6.创建消费者
            MessageConsumer consumer = session.createConsumer(destination);

            // 7.配置监听
            consumer.setMessageListener(getInstance());

        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    @PreDestroy
    public void destroy() {
        if (con != null)
            try {
                con.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
    }

}