Производительность операторов MySQL Insert в Java: готовые операторы пакетного режима против одной вставки с несколькими значениями

Ответ 1

JDBC - это просто стандарт доступа к базе данных Java SE, предлагающий стандартные интерфейсы, поэтому вы не привязаны к конкретной реализации JDBC. Соединитель MySQL Java (Connector/J) - это реализация интерфейсов JDBC только для баз данных MySQL. По опыту я участвую в проекте, который использует огромное количество данных с использованием MySQL, и мы в основном предпочитаем MyISAM для данных, которые могут быть сгенерированы: это позволяет добиться гораздо более высокой производительности, теряя транзакции, но, как правило, MyISAM быстрее, но InnoDB является более надежным.

Я задавался вопросом о производительности инструкций INSERT примерно год назад и нашел на своем полке кода следующий старый тестовый код (извините, он немного сложный и немного из вашей области вопросов). В приведенном ниже коде представлены примеры 4 способов вставки тестовых данных:

• single INSERT s;
• пакет INSERT s;
• ручной объем INSERT (никогда не используйте его - это опасно);
• и, наконец, подготовленный объем INSERT).

Он использует TestNG в качестве бегуна и использует некоторые пользовательские устаревшие коды кода, например:

• метод runWithConnection() - гарантирует, что соединение будет закрыто или возвращено в пул соединений после выполнения обратного вызова (но в приведенном ниже коде используется ненадежная стратегия закрытия оператора - даже без try/finally для сокращения кода);
• IUnsafeIn<T, E extends Throwable> - пользовательский интерфейс обратного вызова для методов, принимающих один параметр, но потенциально бросающий исключение типа E, например: void handle(T argument) throws E;.

package test;

import test.IUnsafeIn;

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

import static java.lang.String.format;
import static java.lang.String.valueOf;
import static java.lang.System.currentTimeMillis;

import core.SqlBaseTest;
import org.testng.annotations.AfterSuite;
import org.testng.annotations.BeforeSuite;
import org.testng.annotations.BeforeTest;
import org.testng.annotations.Test;

public final class InsertVsBatchInsertTest extends SqlBaseTest {

    private static final int ITERATION_COUNT = 3000;

    private static final String CREATE_TABLE_QUERY = "CREATE TABLE IF NOT EXISTS ttt1 (c1 INTEGER, c2 FLOAT, c3 VARCHAR(5)) ENGINE = InnoDB";
    private static final String DROP_TABLE_QUERY = "DROP TABLE ttt1";
    private static final String CLEAR_TABLE_QUERY = "DELETE FROM ttt1";

    private static void withinTimer(String name, Runnable runnable) {
        final long start = currentTimeMillis();
        runnable.run();
        logStdOutF("%20s: %d ms", name, currentTimeMillis() - start);
    }

    @BeforeSuite
    public void createTable() {
        runWithConnection(new IUnsafeIn<Connection, SQLException>() {
            @Override
            public void handle(Connection connection) throws SQLException {
                final PreparedStatement statement = connection.prepareStatement(CREATE_TABLE_QUERY);
                statement.execute();
                statement.close();
            }
        });
    }

    @AfterSuite
    public void dropTable() {
        runWithConnection(new IUnsafeIn<Connection, SQLException>() {
            @Override
            public void handle(Connection connection) throws SQLException {
                final PreparedStatement statement = connection.prepareStatement(DROP_TABLE_QUERY);
                statement.execute();
                statement.close();
            }
        });
    }

    @BeforeTest
    public void clearTestTable() {
        runWithConnection(new IUnsafeIn<Connection, SQLException>() {
            @Override
            public void handle(Connection connection) throws SQLException {
                final PreparedStatement statement = connection.prepareStatement(CLEAR_TABLE_QUERY);
                statement.execute();
                statement.close();
            }
        });
    }

    @Test
    public void run1SingleInserts() {
        withinTimer("Single inserts", new Runnable() {
            @Override
            public void run() {
                runWithConnection(new IUnsafeIn<Connection, SQLException>() {
                    @Override
                    public void handle(Connection connection) throws SQLException {
                        for ( int i = 0; i < ITERATION_COUNT; i++ ) {
                            final PreparedStatement statement = connection.prepareStatement("INSERT INTO ttt1 (c1, c2, c3) VALUES (?, ?, ?)");
                            statement.setInt(1, i);
                            statement.setFloat(2, i);
                            statement.setString(3, valueOf(i));
                            statement.execute();
                            statement.close();
                        }
                    }
                });
            }
        });
    }

    @Test
    public void run2BatchInsert() {
        withinTimer("Batch insert", new Runnable() {
            @Override
            public void run() {
                runWithConnection(new IUnsafeIn<Connection, SQLException>() {
                    @Override
                    public void handle(Connection connection) throws SQLException {
                        final PreparedStatement statement = connection.prepareStatement("INSERT INTO ttt1 (c1, c2, c3) VALUES (?, ?, ?)");
                        for ( int i = 0; i < ITERATION_COUNT; i++ ) {
                            statement.setInt(1, i);
                            statement.setFloat(2, i);
                            statement.setString(3, valueOf(i));
                            statement.addBatch();
                        }
                        statement.executeBatch();
                        statement.close();
                    }
                });
            }
        });
    }

    @Test
    public void run3DirtyBulkInsert() {
        withinTimer("Dirty bulk insert", new Runnable() {
            @Override
            public void run() {
                runWithConnection(new IUnsafeIn<Connection, SQLException>() {
                    @Override
                    public void handle(Connection connection) throws SQLException {
                        final StringBuilder builder = new StringBuilder("INSERT INTO ttt1 (c1, c2, c3) VALUES ");
                        for ( int i = 0; i < ITERATION_COUNT; i++ ) {
                            if ( i != 0 ) {
                                builder.append(",");
                            }
                            builder.append(format("(%s, %s, '%s')", i, i, i));
                        }
                        final String query = builder.toString();
                        final PreparedStatement statement = connection.prepareStatement(query);
                        statement.execute();
                        statement.close();
                    }
                });
            }
        });
    }

    @Test
    public void run4SafeBulkInsert() {
        withinTimer("Safe bulk insert", new Runnable() {
            @Override
            public void run() {
                runWithConnection(new IUnsafeIn<Connection, SQLException>() {
                    private String getInsertPlaceholders(int placeholderCount) {
                        final StringBuilder builder = new StringBuilder("(");
                        for ( int i = 0; i < placeholderCount; i++ ) {
                            if ( i != 0 ) {
                                builder.append(",");
                            }
                            builder.append("?");
                        }
                        return builder.append(")").toString();
                    }

                    @SuppressWarnings("AssignmentToForLoopParameter")
                    @Override
                    public void handle(Connection connection) throws SQLException {
                        final int columnCount = 3;
                        final StringBuilder builder = new StringBuilder("INSERT INTO ttt1 (c1, c2, c3) VALUES ");
                        final String placeholders = getInsertPlaceholders(columnCount);
                        for ( int i = 0; i < ITERATION_COUNT; i++ ) {
                            if ( i != 0 ) {
                                builder.append(",");
                            }
                            builder.append(placeholders);
                        }
                        final int maxParameterIndex = ITERATION_COUNT * columnCount;
                        final String query = builder.toString();
                        final PreparedStatement statement = connection.prepareStatement(query);
                        int valueIndex = 0;
                        for ( int parameterIndex = 1; parameterIndex <= maxParameterIndex; valueIndex++ ) {
                            statement.setObject(parameterIndex++, valueIndex);
                            statement.setObject(parameterIndex++, valueIndex);
                            statement.setObject(parameterIndex++, valueIndex);
                        }
                        statement.execute();
                        statement.close();
                    }
                });
            }
        });
    }

}

Взгляните на методы, аннотированные аннотацией @Test: они фактически выполняют операторы INSERT. Также обратите внимание на константу CREATE_TABLE_QUERY: в исходном коде он использует InnoDB, производя следующие результаты на моей машине с установленным MySQL 5.5 (MySQL Connector/J 5.1.12):

InnoDB
Single inserts: 74148 ms
Batch insert: 84370 ms
Dirty bulk insert: 178 ms
Safe bulk insert: 118 ms

Если вы измените CREATE_TABLE_QUERY InnoDB на MyISAM, вы увидите значительное увеличение производительности:

MyISAM
Single inserts: 604 ms
Batch insert: 447 ms
Dirty bulk insert: 63 ms
Safe bulk insert: 26 ms

Надеюсь, что это поможет.

UPD:

Для 4-го пути вы должны правильно настроить max_allowed_packet в mysql.ini (раздел [mysqld]), чтобы быть достаточно большим, чтобы поддерживать действительно большие пакеты.

Ответ 2

Я знаю, что этот поток довольно старый, но я просто подумал, что хочу упомянуть, что если вы добавите "rewriteBatchedStatements = true" в URL-адрес jdbc при использовании mysql, это может привести к огромному увеличению производительности при использовании пакетных операторов.

Ответ 3

У вас есть триггеры в любой из затронутых таблиц? Если нет, то 600 вставок в секунду не выглядят много.

Функциональность пакетной вставки из JDBC будет выдавать один и тот же оператор несколько раз в одной транзакции, тогда как многозначный SQL сжимает все значения в одном выражении. В случае многозначного оператора вам придется динамически строить вставку SQL, и это может быть накладными расходами с точки зрения большего количества кода, большего объема памяти, механизма защиты SQL Injection и т.д. Сначала попробуйте обычную функциональность пакета, для вашей рабочей нагрузки, не должно быть проблемой.

Если вы не получаете данные в партиях, тогда рассмотрите их до ввода. Мы используем Очередь в отдельном потоке, чтобы реализовать устройство Продюсер-Потребитель. В этом случае мы удерживаем вставки до истечения определенного времени или размер очереди пересек порог.

Если вы хотите, чтобы продюсер был уведомлен об успешной вставке, требуется дополнительная сантехника.

Иногда просто блокировка на потоке может быть более прямой и практичной.

if(System.currentTimeMills()-lastInsertTime>TIME_THRESHOLD || queue.size()>SIZE_THRESHOLD) {
    lastInsertTime=System.currentTimeMills();
    // Insert logic
    } else {
    // Do nothing OR sleep for some time OR retry after some time. 
    }