dbconnectionmanager.java

封装了SQL、Socket、WAP、MIME等功能的通用组件

package org.lazybug.sql;

/**
 * <p>Title: </p>
 * <p>Description: </p>
 * <p>Copyright: Copyright (c) 2003</p>
 * <p>Company: </p>
 * @author not attributable
 * @version 1.0
 */
import org.lazybug.util.*;
import java.sql.*;
import java.util.*;

public class DBConnectionManager extends QueneListener
{
    private static DBConnectionManager manager = null;
    private static DBConnectionPool pool = null;
    public static boolean bIsConDbs = false;

    public static DBConnectionManager getInstance()
    {
        if( manager == null )
        {
            initialize();
        }
        return manager;
    }

    private DBConnectionManager()
    {
        start();
        thread.setPriority(Thread.MIN_PRIORITY);
        bIsConDbs = true;
    }

    private static void initialize()
    {
        if( pool == null )
        {
            try
            {
                String db_url = "jdbc:mysql://"+ConfigUtil.getString(
                    "server_host")+"/"+ConfigUtil.getString("db_name");
                pool = new DBConnectionPool(ConfigUtil.getString("db_driver"),
                                            db_url,
                                            ConfigUtil.getString("db_user"),
                                            ConfigUtil.getString("db_password"));
                manager = new DBConnectionManager();
            }
            catch(Exception e)
            {
                bIsConDbs = false;
                Log.logError(manager, "Failed to connect database for "+e);
            }
        }
    }

    public static final boolean isConDbs()
    {
        return bIsConDbs;
    }
    public synchronized void close()
    {
        this.stop();
        this.notify();
        Log.logMessage(this, "释放数据库连接之前需要同步数据，请稍等……");
    }
    /**
     * 执行数据库更新操作
     * @param szSQL
     */
    public void executeUpdate(String szSQL)
    {
        DBConnection dbconn = this.pool.getConnection();
        if( dbconn != null )
        {
            Connection conn = dbconn.get();
            try
            {
                Statement stmt = conn.createStatement();
                stmt.executeUpdate(szSQL);
                stmt.close();
            }
            catch (java.sql.SQLException e)
            {
                Log.logError(this, "执行SQL语句失败！" + e);
            }
            finally
            {
                this.pool.freeConnection(dbconn);
            }
        }
    }

    public int getCount(String szSQL){
        int count = 0;
        DBConnection dbconn = this.pool.getConnection();
        if( dbconn != null )
        {
            Connection conn = dbconn.get();
            try
            {
                Statement stmt = conn.createStatement();
                ResultSet rs = stmt.executeQuery(szSQL);
                if( rs.next() ){
                    count = rs.getInt(1);
                }
                else{
                    count = -1;
                }
                rs.close();
                stmt.close();
            }
            catch (java.sql.SQLException e){
                Log.getInstance().write("查询记录个数失败:"+e);
                count = -1;
            }
            finally
            {
                this.pool.freeConnection(dbconn);
            }
        }
        return count;
    }
    /**
     * 获取数据库连接
     */
    public static DBConnection getConnection()
    {
        if( pool == null )
            initialize();
        if( pool == null ) return null;
        DBConnection dbconn = pool.getConnection(1000);
        return dbconn;
    }
    /**
     * 释放数据库连接
     */
    public static void freeConnection(DBConnection conn)
    {
        pool.freeConnection(conn);
    }

    /**
     * 插入执行数据库操作的SQl语句
     */
    public static void postSQL(String szSQL)
    {
        manager.post(szSQL);
    }

    /**
     * 执行平台提交的数据库操作语句
     */
    public void run()
    {
        this.remark = "[数据库管理器]";
        int upperlimit = 1024;//批处理上限条数
        int rows = 0;
        Suspender suspender = new Suspender(this.thread, 1, 300);//暂停器
        Log.logMessage(manager, "DBS update listener service is start!");

        while( isRunning || !this.quene.isEmpty() )
        {
            rows = 0;//批处理条数计数器
            synchronized( this ){
                try{
                    this.isBusy = false;
                    if( this.quene.isEmpty() ) wait();
                    this.isBusy = true;
                }catch(InterruptedException e){}
            }

            DBConnection dbconn = pool.getConnection(Tools.MILLI_OF_MINUTE);
            if( dbconn != null )
            {
                Connection conn = dbconn.get();
                try
                {
                    Object szSQL = null;
                    Statement statement = conn.createStatement();
                    while ( (szSQL = this.peek()) != null)
                    {
                        rows += 1;
                        statement.addBatch(szSQL.toString());
                        if (rows >= upperlimit)
                            break; //如果超过一次批处理上限条数就退出循环
                    }
                    if( rows > 0 )
                    {
                        int rowUpdated = statement.executeBatch().length;
                        statement.clearBatch();
                    }
                    statement.close();

                    if (rows == upperlimit)
                    {
                        suspender.suspend(); //如果超过上限条数才挂起
                    }
                }
                catch (SQLException ex)
                {
                    Log.getInstance().write(ex.getMessage());
                }
                finally
                {
                    pool.freeConnection(dbconn);
                }
            }
        }
        this.isBusy = false;
        pool.close();
        Log.logMessage(manager, "DBS update listener service is close!");
    }

    public String toString()
    {
        StringBuffer sb = new StringBuffer();
        sb.append("[硬盘数据库]待处理的SQL语句[");
        sb.append(this.quene.size());
        sb.append("];");
        sb.append(this.pool.toString());
        return sb.toString();
    }
}