MysqlとJavaソケットの並列ストレステスト

Question

私はJavaのTCPソケットアプリケーションを持っています。 TcpコミュニケータはGPSの位置をデコードし、そのデータをデータベースに挿入します。挿入する前にいくつかの選択と更新を行いますが、準備されたステートメントを使用してすべて行います。現在、TCPコミュニケータの1つのスレッドが1つのデバイス要求を処理します。スレッドを作成した直後に、プールから1つの接続を取得します。デコード後、GPSデータは、各データに対して複数の選択、更新、挿入を実行する場所です。デバイスの数が増えているので、MySQLデータベースへの同時接続数も増えています。私は以下のようなシミュレーションとストレステストを行っています。問題はこれが順次テストであることですが、実際の環境ではデバイスは並行して動作します。どのようにmysqlとjavaの両方のために近い実際のストレス状況を達成するためにいくつの挿入をmysqlが2番目に取ることができますか？

public class stress1 extends Thread { 
    public static void main(String[] argv) { 
     try { 
      for (int i = 0; i < 5000; i++) { 
      Socket socket = new Socket("192.168.2.102", 8000); 
      PrintWriter out = new PrintWriter(socket.getOutputStream(), true); 
      BufferedReader in = new BufferedReader(new InputStreamReader(
      socket.getInputStream())); 
      out.println("$A12345,30061104075130528955N10024852E000068*03A1*"); 
      System.out.println(in.readLine() + i); 
      out.close(); 
      in.close(); 
      socket.close(); 
      } 
     } catch (Exception e) { 
      e.printStackTrace(); 
     } 
     } 
    } 

私のサーバーソケットの外観は次のとおりです。

public class comm8888 { 
    HikariDataSource connectionPool = null; 
    private Socket receivedSocketConn1; 
    ConnectionHandler(Socket receivedSocketConn1) { 
     this.receivedSocketConn1=receivedSocketConn1; 
    } 
    Connection dbconn = null; 
    public void run() { // etc 
    DataOutputStream w = null; 
    DataInputStream r = null; 
    String message=""; 
    receivedSocketConn1.setSoTimeout(60000); 
     dbconn = connectionPool.getConnection(); 
    dbconn.setAutoCommit(false); 
    try { 
     w = new DataOutputStream(new BufferedOutputStream(receivedSocketConn1.getOutputStream())); 
     r = new DataInputStream(new BufferedInputStream(receivedSocketConn1.getInputStream())); 
     while ((m=r.read()) != -1){ 
      //multiple prepared based sql select,update and insert here. 
     } 
    } 
    finally{ 
     try { 
      if (dbconn != null) { 
       dbconn.close(); 
      } 
     } 
     catch(SQLException ex){ 
      ex.printStackTrace(); 
     } 
     try{ 
      if (w != null){ 
       w.close(); 
       r.close(); 
       receivedSocketConn1.close(); 
      } 
     } 
     catch(IOException ex){ 
      ex.printStackTrace(System.out); 
     } 
     } 
    } 
} 
    public static void main(String[] args) { 
     new comm8888(); 
    } 
    comm8888() { 
     try { 
      HikariConfig config = new HikariConfig(); 
       config.setJdbcUrl("jdbc:mysql://localhost:3306/testdata"); 
       config.setUsername("****"); 
       config.setPassword("****");  
       config.setMaximumPoolSize(20);  
      connectionPool = new HikariDataSource(config); // setup the connection pool   
     } 
      catch (Exception e) { 
       e.printStackTrace(System.out); 
     } 
      try 
      { 
        final ServerSocket serverSocketConn = new ServerSocket(8888);     
        while (true){ 
          try { 
            Socket socketConn1 = serverSocketConn.accept(); 
            new Thread(new ConnectionHandler(socketConn1)).start();      
          } 
          catch(Exception e){ 
           e.printStackTrace(System.out); 
          } 
         } 
      } 
      catch (Exception e) { 
      e.printStackTrace(System.out); 
      } 
    } 
} 

Answer 1

解決策はうまく拡張されません。私はいくつかのクラスのオブジェクトに行われる作業（デバイスのGPS位置）をカプセル化し、各作業単位をキューに入れます。最後に、1つのスレッドがすべての作業を順番に処理し、一度に1つの要求を処理できます。

スレッドが追いつくことができず、キューがいっぱいになると、キューからジョブを処理するためにさらに多くのワーカーを追加することで、スケーリングが非常に簡単になります。 （または、MySQLの1つのインスタンスがすべての挿入を処理できない場合は、データの水平シャーディングや複数のMySQLインスタンスの追加なども試すことができます）。

これは、いくつかのサンプルコードです：

import java.sql.Connection; 
import java.util.concurrent.ArrayBlockingQueue; 
import java.util.concurrent.BlockingQueue; 


public class QueueingDatabaseProcessingExample { 

    public static void main(String[] args) throws InterruptedException { 
     QueueingDatabaseProcessingExample a = new QueueingDatabaseProcessingExample(); 
     a.doTheWork(); 
    } 

    private void doTheWork() throws InterruptedException { 

     BlockingQueue<TcpCommunicatorUnitOfWork> blockingQueue = new ArrayBlockingQueue(1000); 

     // add work to queue as needed 
     blockingQueue.put(new TcpCommunicatorUnitOfWork("device id", 40.7143528, -74.0059731, 10)); // blocks if queue is full 


     Connection connection; 

     // get connection to the database from database pool 

     // process requests one by one sequentially 
     while (true) { 
      TcpCommunicatorUnitOfWork tcpCommunicatorUnitOfWork = blockingQueue.take(); // blocks if queue is empty 

      proccess(tcpCommunicatorUnitOfWork); 
     } 
    } 

    private void proccess(TcpCommunicatorUnitOfWork tcpCommunicatorUnitOfWork) { 
     // do queries, inserts, deletes to database 
    } 

} 


/** 
* this class should have all the information needed to query/update the database 
*/ 
class TcpCommunicatorUnitOfWork { 

    private final String deviceId; 
    private final double latitude; 
    private final double longitude; 
    private final int seaLevel; 

    public TcpCommunicatorUnitOfWork(String deviceId, double latitude, double longitude, int seaLevel) { 
     this.deviceId = deviceId; 
     this.latitude = latitude; 
     this.longitude = longitude; 
     this.seaLevel = seaLevel; 
    } 


}