spark streaming将处理结果存入mysql中（使用c3p0连接池）

1、c3p0相应的架包导入工程中

将以下四个架包导入工程，

　　主要有三个架包：c3p0-0.9.5.2.jar

                                c3p0-oracle-thin-extras-0.9.5.2.jar

                                mchange-commons-java-0.9.5.2.jar

　　记得一定要导入mysql-connector-java-5.1.26-bin.jar架包，这是连接mysql的驱动程序。

2、创建进程池

　　主要有三个类：

　　　C3p0Utils创建连接以及关闭连接；

　　   DBUtils插入更新等执行语句；

　　　DBUtil_BO数据库连接对像；

　　　App测试；

import com.mchange.v2.c3p0.ComboPooledDataSource;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;

/**
 * Created by lala on 2017/6/29.
 */

public class C3p0Utils {

    static org.apache.log4j.Logger logger=org.apache.log4j.Logger.getLogger(C3p0Utils.class.getName());

    //通过标识名来创建相应连接池
    static ComboPooledDataSource dataSource=new ComboPooledDataSource("mysql");
    //从连接池中取用一个连接
    public static Connection getConnection(){
        try {
             String url="jdbc:mysql://localhost:3306/mysql";
            //定义连接数据的用户名
            String user="root";
            //定义连接数据库的密码
             String password="";
            return dataSource.getConnection();

        } catch (Exception e) {
            logger.error("Exception in C3p0Utils!", e);
           System.out.println("数据库连接出错!"+e);
            return null;
        }
    }
    //释放连接回连接池
    public static void close(Connection conn, PreparedStatement pst, ResultSet rs){
        if(rs!=null){
            try {
                rs.close();
            } catch (SQLException e) {
                logger.error("Exception in C3p0Utils!", e);
                System.out.println("数据库连接关闭出错!"+e);
            }
        }
        if(pst!=null){
            try {
                pst.close();
            } catch (SQLException e) {
                logger.error("Exception in C3p0Utils!", e);
                System.out.println("数据库连接关闭出错!"+ e);
            }
        }

        if(conn!=null){
            try {
                conn.close();
            } catch (SQLException e) {
                logger.error("Exception in C3p0Utils!", e);
                System.out.println("数据库连接关闭出错!"+ e);
            }
        }
    }

}

DBUtils插入更新等执行语句：

public class DBUtils {

    static org.apache.log4j.Logger logger=org.apache.log4j.Logger.getLogger(DBUtils.class.getName());


    private static void realseSource(Connection _conn, PreparedStatement _st, ResultSet _rs){
        C3p0Utils.close(_conn,_st,_rs);
    }

    public static void realseSource(DBUtil_BO _vo){
        if(_vo!=null){
            realseSource(_vo.conn, _vo.st, _vo.rs);
        }
    }
    //注意：查询操作完成后，因为还需提取结果集中信息，所以仍保持连接，在结果集使用完后才通过DBUtils.realseSource()手动释放连接
    public static void executeQuery(DBUtil_BO vo)
    {
        try{
            vo.rs = vo.st.executeQuery();
        }catch (SQLException e){
            realseSource(vo);

            logger.error("SQL语法有误: ",e);
           System.out.println("err.user.dao.jdbc"+e);
        }
    }

    //而update操作完成后就可以直接释放连接了，所以在方法末尾直接调用了realseSourse()
    public static  void executeUpdate(DBUtil_BO vo)
    {

        Connection conn = vo.conn;
        PreparedStatement st = vo.st;
        try {
            st.executeUpdate();
        } catch (SQLException e) {
            realseSource(conn, st, null);
            logger.error(" SQL语法有误: ",e);
            System.out.println("err.user.dao.jdbc"+e);
        }
        realseSource(conn, st,null );
    }
}

DBUtil_BO数据库连接对像：

package org.test.maven;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

/**
 * Created by lala on 2017/6/29.
 */
public class DBUtil_BO {

    public Connection conn = null;
    public PreparedStatement st = null;
    public ResultSet rs = null;
    public DBUtil_BO() {
        super();
    }

}

Obj方便插入的某一对象(scala)

package org.test.maven

/**
  * Created by lala on 2017/7/1.
  */
class Obj {
  private var usr = -1
  private var name = ""

  def setUsr(usr:Int)={
    this.usr=usr
  }

  def  setName(name:String)={
    this.name=name
  }
  def getUsr()=this.usr
  def getName()=this.name
}

App测试：

package org.test.maven;

import java.sql.SQLException;
import org.apache.log4j.Logger;

/**
 * Hello world!
 *
 */
public class App 
{

    private static Logger logger = Logger.getLogger(App.class);

    public static void main( String[] args ) throws SQLException {
        System.out.println( "Hello World!" );
        DBUtil_BO dbBo = new DBUtil_BO();
        dbBo.conn=C3p0Utils.getConnection();//取用一个连接
        String sql = "select id from usr where usr_name = ? ";
        try{
            dbBo.st=dbBo.conn.prepareStatement(sql);//预处理sql语句
        }catch (SQLException e){
            logger.error("查询预处理出错!",e);
            System.out.println("查询预处理出错!"+e);
        }
        Obj ob=new Obj();
        ob.setName("bgngh");
        System.out.println(ob.getName());
        dbBo.st.setString(1,ob.getName());
        DBUtils.executeQuery(dbBo);
        String name = null;
        //从dbBo类提取操作结果
        int id = 0;
        if (dbBo.rs.next()) {
            try{

                 id = dbBo.rs.getInt("id");
                 id = 5;
            }catch(SQLException e){
                logger.error("查询返回结果出错!",e);
            }

        }
        System.out.println(id);
//结果集遍历完了，手动释放连接回连接池
        DBUtils.realseSource(dbBo);
    }
}