java - 可以使用 JAVA sql 打开两个到不同远程 MySQL 服务器的持久连接吗？

Question

因此，我尝试运行一段代码，通过引用位于不同 MYSQL 服务器中的表，在远程 MySQL 服务器中创建一个表。我试图在其中创建表的服务器有空间限制，并且正在引用的表非常大，因此必须将它们保存在不同的远程服务器上。

我正在尝试找到一种方法来同时建立与两个数据库的持久连接(使用 JDBC 库)，这样我就不必一直缓冲少量的数据行......我想要能够直接引用数据。

例如数据库 A 包含我正在引用的数据，数据库 B 是我创建新表的地方。假设我在数据库 A 中引用的表是 1,000,000 行。而不是打开与数据库 A 的连接，缓冲 10,000 行，关闭连接，打开与数据库 B 的连接，写入该数据库，删除我的缓冲区，然后重复...

我只想与数据库 A 建立持久连接，这样每次对数据库 B 的写入都可以引用数据库 A 中的数据。

这可能吗？我尝试了几种方法(主要是通过创建仅在连接中断时刷新的新连接对象)，但我似乎无法让这个想法发挥作用。

有人用 JDBC 做过类似的事情吗？如果是这样，如果您能为我指出正确的方向，或者告诉我您是如何让它发挥作用的，我们将不胜感激。

Answer 1

您可以在数据库 A 中创建数据，然后通过复制将其复制到数据库 B。

或者，听起来您正在实现某种队列。我曾经用 Java 构建了一个数据复制程序，它使用了 Queue 接口(interface)的内置实现。我有一个从数据库 A 读取数据并填充队列的线程，以及一个从队列读取数据并写入数据库 B 的线程。如果有任何用处，我可以尝试挖掘我使用的类吗？

编辑:

这是代码，为发布做了一些调整。我没有包括配置类，但它应该让您了解如何使用队列类；

package test;

import java.io.File;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * This class implements a JDBC bridge between databases, allowing data to be
 * copied from one place to another.
 * <p>This implementation is threaded, as it uses a {@link BlockingQueue} to pass
 * data between a producer and a consumer.
 */
public class DBBridge {

    public static void main( String[] args ) {

        Adaptor fromAdaptor = null;
        Adaptor toAdaptor = null;

        BridgeConfig config = null;

        try {
            /* BridgeConfig is essentially a wrapper around the Simple XML serialisation library.
             * http://simple.sourceforge.net/
             */
            config = BridgeConfig.loadConfig( new File( "db-bridge.xml" ) );
        }
        catch ( Exception e ) {
            System.err.println( "Failed to read or parse db-bridge.xml: " + e.getLocalizedMessage() );
            System.exit( 1 );
        }

        BlockingQueue<Object> b = new ArrayBlockingQueue<Object>( config.getQueueSize() );

        try {
            HashMap<String, DatabaseConfig> dbs = config.getDbs();

            System.err.println( "Configured DBs" );

            final String sourceName = config.getSource();
            final String destinationName = config.getDestination();

            if ( !dbs.containsKey( sourceName ) ) {
                System.err.println( sourceName + " is not a configured database connection" );
                System.exit( 1 );
            }

            if ( !dbs.containsKey( destinationName ) ) {
                System.err.println( destinationName + " is not a configured database connection" );
                System.exit( 1 );
            }

            DatabaseConfig sourceConfig = dbs.get( sourceName );
            DatabaseConfig destinationConfig = dbs.get( destinationName );

            try {
                /*
                 * Both adaptors must be created before attempting a connection,
                 * as otherwise I've seen DriverManager pick the wrong driver!
                 */
                fromAdaptor = AdaptorFactory.buildAdaptor( sourceConfig, sourceConfig );
                toAdaptor = AdaptorFactory.buildAdaptor( destinationConfig, destinationConfig );

                System.err.println( "Connecting to " + sourceName );
                fromAdaptor.connect();

                System.err.println( "Connecting to " + destinationName );
                toAdaptor.connect();

                /* We'll send our updates to the destination explicitly */
                toAdaptor.getConn().setAutoCommit( false );
            }
            catch ( SQLException e ) {
                System.err.println();
                System.err.println( "Failed to create and configure adaptors" );
                e.printStackTrace();
                System.exit( 1 );
            }
            catch ( ClassNotFoundException e ) {
                System.err.println( "Failed to load JDBC driver due to error: " + e.getLocalizedMessage() );
                System.exit( 1 );
            }

            DataProducer producer = null;
            DataConsumer consumer = null;

            try {
                producer = new DataProducer( config, fromAdaptor, b );
                consumer = new DataConsumer( config, toAdaptor, b );
            }
            catch ( SQLException e ) {
                System.err.println();
                System.err.println( "Failed to create and configure data producer or consumer" );
                e.printStackTrace();
                System.exit( 1 );
            }

            consumer.start();
            producer.start();
        }
        catch ( Exception e ) {
            e.printStackTrace();
        }
    }

    public static class DataProducer extends DataLogger {

        private BridgeConfig config;
        private Adaptor adaptor;
        private BlockingQueue<Object> queue;


        public DataProducer(BridgeConfig c, Adaptor a, BlockingQueue<Object> bq) {
            super( "Producer" );
            this.config = c;
            this.adaptor = a;
            this.queue = bq;
        }


        @Override
        public void run() {
            /* The tables to copy are listed in BridgeConfig */
            for ( Table table : this.config.getManifest() ) {

                PreparedStatement stmt = null;
                ResultSet rs = null;

                try {
                    String sql = table.buildSourceSelect();
                    this.log( "executing: " + sql );
                    stmt = this.adaptor.getConn().prepareStatement( sql );

                    stmt.execute();

                    rs = stmt.getResultSet();

                    ResultSetMetaData meta = rs.getMetaData();

                    /* Notify consumer that a new table is to be processed */
                    this.queue.put( table );
                    this.queue.put( meta );

                    final int columnCount = meta.getColumnCount();

                    while ( rs.next() ) {
                        ArrayList<Object> a = new ArrayList<Object>( columnCount );

                        for ( int i = 0; i < columnCount; i++ ) {
                            a.add( rs.getObject( i + 1 ) );
                        }

                        this.queue.put( a );
                    }
                }
                catch ( InterruptedException ex ) {
                    ex.printStackTrace();
                }
                catch ( SQLException e ) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }

                try {
                    /* refresh the connection */
                    /* Can't remember why I can this line - maybe the other
                     * end kept closing the connection. */
                    this.adaptor.reconnect();
                }
                catch ( SQLException e ) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }

            try {
                /* Use an object of a specific type to "poison" the queue
                 * and instruct the consumer to terminate. */
                this.log( "putting finished object into queue" );
                this.queue.put( new QueueFinished() );

                this.adaptor.close();
            }
            catch ( InterruptedException e ) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            catch ( SQLException e ) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }

    }

    /* Superclass for producer and consumer */
    public static abstract class DataLogger extends Thread {

        private String prefix;


        public DataLogger(String p) {
            this.prefix = p;
        }


        protected void log( String s ) {
            System.err.printf( "%d %s: %s%n", System.currentTimeMillis(), this.prefix, s );
        }


        protected void log() {
            System.err.println();
        }
    }

    public static class DataConsumer extends DataLogger {

        private BridgeConfig config;
        private Adaptor adaptor;
        private BlockingQueue<Object> queue;
        private int currentRowNumber = 0;
        private int currentBatchSize = 0;
        private long tableStartTs = -1;


        public DataConsumer(BridgeConfig c, Adaptor a, BlockingQueue<Object> bq) throws SQLException {

            super( "Consumer" );

            this.config = c;
            this.adaptor = a;
            this.queue = bq;

            /* We'll send our updates to the destination explicitly */
            this.adaptor.getConn().setAutoCommit( false );
        }


        public void printThroughput() {
            double duration = ( System.currentTimeMillis() - this.tableStartTs ) / 1000.0;
            long rowsPerSec = Math.round( this.currentRowNumber / duration );
            this.log( String.format( "%d rows processed, %d rows/s", this.currentRowNumber, rowsPerSec ) );
        }


        @Override
        public void run() {

            this.log( "running" );

            Table currentTable = null;
            ResultSetMetaData meta = null;

            int columnCount = -1;

            PreparedStatement stmt = null;

            while ( true ) {
                try {
                    Object o = this.queue.take();

                    if ( o instanceof Table ) {
                        currentTable = (Table) o;

                        this.log( "processing" + currentTable );

                        if ( this.currentBatchSize > 0 ) {
                            /* Commit outstanding rows from previous table */

                            this.adaptor.getConn().commit();

                            this.printThroughput();
                            this.currentBatchSize = 0;
                        }

                        /* refresh the connection */
                        this.adaptor.reconnect();
                        this.adaptor.getConn().setAutoCommit( false );

                        /*
                         * Arguably, there's no need to flush the commit buffer
                         * after every table, but I like it because it feels
                         * tidy.
                         */
                        this.currentBatchSize = 0;
                        this.currentRowNumber = 0;

                        if ( currentTable.isTruncate() ) {
                            this.log( "truncating " + currentTable );
                            stmt = this.adaptor.getConn().prepareStatement( "TRUNCATE TABLE " + currentTable );
                            stmt.execute();
                        }

                        this.tableStartTs = System.currentTimeMillis();
                    }
                    else if ( o instanceof ResultSetMetaData ) {

                        this.log( "received metadata for " + currentTable );

                        meta = (ResultSetMetaData) o;
                        columnCount = meta.getColumnCount();

                        String sql = currentTable.buildDestinationInsert( columnCount );
                        stmt = this.adaptor.getConn().prepareStatement( sql );
                    }
                    else if ( o instanceof ArrayList ) {

                        ArrayList<?> a = (ArrayList<?>) o;

                        /* One counter for ArrayList access, one for JDBC access */
                        for ( int i = 0, j = 1; i < columnCount; i++, j++ ) {

                            try {
                                stmt.setObject( j, a.get( i ), meta.getColumnType( j ) );
                            }
                            catch ( SQLException e ) {
                                /* Sometimes data in a shonky remote system
                                 * is rejected by a more sane destination
                                 * system. Translate this data into
                                 * something that will fit. */
                                if ( e.getMessage().contains( "Only dates between" ) ) {

                                    if ( meta.isNullable( j ) == ResultSetMetaData.columnNullable ) {
                                        this.log( "Casting bad data to null: " + a.get( i ) );
                                        stmt.setObject( j, null, meta.getColumnType( j ) );
                                    }
                                    else {
                                        this.log( "Casting bad data to 0000-01-01: " + a.get( i ) );
                                        stmt.setObject( j, new java.sql.Date( -64376208000L ), meta.getColumnType( j ) );
                                    }
                                }
                                else {
                                    throw e;
                                }
                            }
                        }

                        stmt.execute();

                        this.currentBatchSize++;
                        this.currentRowNumber++;

                        if ( this.currentBatchSize == this.config.getBatchSize() ) {
                            /*
                             * We've reached our non-committed limit. Send the
                             * requests to the destination server.
                             */

                            this.adaptor.getConn().commit();

                            this.printThroughput();
                            this.currentBatchSize = 0;
                        }
                    }
                    else if ( o instanceof QueueFinished ) {
                        if ( this.currentBatchSize > 0 ) {
                            /* Commit outstanding rows from previous table */

                            this.adaptor.getConn().commit();

                            this.printThroughput();

                            this.log();
                            this.log( "completed" );
                        }

                        /* Exit while loop */
                        break;
                    }
                    else {
                        throw new RuntimeException( "Unexpected obeject in queue: " + o.getClass() );
                    }
                }
                catch ( InterruptedException e ) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
                catch ( SQLException e ) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }

            try {
                this.adaptor.close();
            }
            catch ( SQLException e ) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }

    public static final class QueueFinished {
        /*
         * This only exists as a completely type-safe value in "instanceof"
         * expressions
         */
    }
}