Mysql und Java-Buchse parallelen Stresstest

Question

Ich habe eine Java-TCP-Socket-Anwendung. Tcp communicator dekodiert den GPS-Standort und fügt diese Daten in die Datenbank ein. Vor dem Einfügen führen wir einige Selects und Updates durch, aber alles, was ich mache, benutze vorbereitete Anweisungen. Momentan bedient ein Thread des TCP-Communicators eine Geräteanforderung. Sofort nach dem Erstellen des Threads erhalten wir eine Verbindung aus dem Pool. Nach der Dekodierung der GPS-Daten führen wir die Mehrfachauswahl, Aktualisierung und Einfügung für jede Daten durch. Da die Anzahl der Geräte steigt, steigt auch die Anzahl der gleichzeitigen Verbindungen zu unserer Mysql-Datenbank. Deshalb versuche ich, einen Simulations- und Stresstest wie unten beschrieben durchzuführen. Das Problem ist, dass dies ein sequentieller Test ist, aber in einer realen Umgebung werden die Geräte parallel geschaltet. Wie erreicht man eine nahezu reale Stresssituation für mysql und java, um herauszufinden, wie viele Einfügungen mysql auf Sekunde aufnehmen könnte?

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(); 
     } 
     } 
    } 

Hier ist, wie meine Server-Socket aussieht.

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

Ihre Lösung skaliert nicht gut. Ich würde die zu erledigende Arbeit (GPS-Position eines Geräts) in Objekte einer Klasse einkapseln und dann jede Arbeitseinheit in eine Warteschlange stellen. Schließlich kann ein Thread die gesamte Arbeit sequentiell verarbeiten und jeweils eine Anfrage bearbeiten.

Wenn ein Thread nicht mithalten kann und Ihre Warteschlange voll wird, kann er sehr einfach skaliert werden, indem mehr Worker hinzugefügt werden, um Jobs aus der Warteschlange zu verarbeiten. (Wenn eine Instanz von MySQL nicht alle Einfügungen verarbeiten kann, können Sie beispielsweise auch versuchen, Ihre Daten horizontal zu teilen und mehrere MySQL-Instanzen hinzuzufügen).

dies ist ein Beispielcode:

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; 
    } 


}