Sdílet prostřednictvím

Aplikace Java pro připojení a spuštění příkazů SQL ve službě Azure Cosmos DB for PostgreSQL

  • Článek

PLATÍ PRO: Azure Cosmos DB for PostgreSQL (využívající rozšíření databáze Citus do PostgreSQL)

V tomto rychlém startu se dozvíte, jak se pomocí kódu Java připojit ke clusteru a pomocí příkazů SQL vytvořit tabulku. Potom vložíte data do databáze, budete je dotazovat, aktualizovat a odstraňovat. Kroky v tomto článku předpokládají, že znáte vývoj v Javě a JDBC a začínáte pracovat se službou Azure Cosmos DB for PostgreSQL.

Nastavení projektu a připojení Java

Vytvořte nový projekt Java a konfigurační soubor pro připojení ke službě Azure Cosmos DB for PostgreSQL.

Vytvoření nového projektu v Javě

Pomocí svého oblíbeného integrovaného vývojového prostředí (IDE) vytvořte nový projekt Java s id skupiny test a artifactId crud. Do kořenového adresáře projektu přidejte pom.xml soubor s následujícím obsahem. Tento soubor nakonfiguruje Apache Maven tak, aby používal Javu 8 a nedávný ovladač PostgreSQL pro Javu.

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>test</groupId>
  <artifactId>crud</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <packaging>jar</packaging>

  <name>crud</name>
  <url>http://www.example.com</url>

  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
  </properties>

  <dependencies>
    <dependency>
      <groupId>org.junit.jupiter</groupId>
      <artifactId>junit-jupiter-engine</artifactId>
      <version>5.7.1</version>
      <scope>test</scope>
    </dependency>
    <dependency>
      <groupId>org.postgresql</groupId>
      <artifactId>postgresql</artifactId>
      <version>42.2.12</version>
    </dependency>
    <!-- https://mvnrepository.com/artifact/com.zaxxer/HikariCP -->
    <dependency>
      <groupId>com.zaxxer</groupId>
      <artifactId>HikariCP</artifactId>
      <version>5.0.0</version>
    </dependency>
    <dependency>
      <groupId>org.junit.jupiter</groupId>
      <artifactId>junit-jupiter-params</artifactId>
      <version>5.7.1</version>
      <scope>test</scope>
    </dependency>
  </dependencies>

  <build>
    <plugins>
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-surefire-plugin</artifactId>
        <version>3.0.0-M5</version>
      </plugin>
    </plugins>
  </build>
</project>

Konfigurace připojení k databázi

V souboru src/main/resources/vytvořte soubor application.properties s následujícím obsahem. Nahraďte <cluster> názvem clusteru a nahraďte <heslo> heslem pro správu nebo tokenem Microsoft Entra ID.

driver.class.name=org.postgresql.Driver
db.url=jdbc:postgresql://c-<cluster>.<uniqueID>.postgres.cosmos.azure.com:5432/citus?ssl=true&sslmode=require
db.username=citus
db.password=<password>

Řetězec ?ssl=true&sslmode=require ve db.url vlastnosti říká ovladači JDBC, aby při připojování k databázi používal protokol TLS (Transport Layer Security). Je povinné používat protokol TLS se službou Azure Cosmos DB for PostgreSQL a je dobrým postupem zabezpečení.

Vytváření tabulek

Nakonfigurujte schéma databáze s distribuovanými tabulkami. Připojte se k databázi a vytvořte schéma a tabulky.

Generování schématu databáze

V souboru src/main/resources/vytvořte soubor schema.sql s následujícím obsahem:

DROP TABLE IF EXISTS public.pharmacy;
CREATE TABLE  public.pharmacy(pharmacy_id integer,pharmacy_name text ,city text ,state text ,zip_code integer);
CREATE INDEX idx_pharmacy_id ON public.pharmacy(pharmacy_id);

Distribuce tabulek

Azure Cosmos DB for PostgreSQL poskytuje super výkon distribuce tabulek napříč několika uzly pro zajištění škálovatelnosti. Následující příkaz umožňuje distribuovat tabulku. Další informace o create_distributed_table distribučním sloupci najdete tady.

Poznámka:

Distribuce tabulek umožňuje růst mezi všechny pracovní uzly přidané do clusteru.

Pokud chcete distribuovat tabulky, připojte následující řádek k souboru schema.sql , který jste vytvořili v předchozí části.

select create_distributed_table('public.pharmacy','pharmacy_id');

Připojení k databázi a vytvoření schématu

Dále přidejte kód Java, který používá JDBC k ukládání a načítání dat z clusteru. Kód používá soubory application.properties a schema.sql pro připojení ke clusteru a vytvoření schématu.

  1. Vytvořte DButil.java soubor s následujícím kódem, který obsahuje DButil třídu. Třída DBUtil nastaví fond připojení k PostgreSQL pomocí HikariCP. Tuto třídu použijete pro připojení k PostgreSQL a zahájení dotazování.

    Tip

    Následující ukázkový kód používá fond připojení k vytvoření a správě připojení k PostgreSQL. Sdružování připojení na straně aplikace se důrazně doporučuje, protože:

    • Zajišťuje, že aplikace nevygeneruje příliš mnoho připojení k databázi, a proto se vyhne překročení limitů připojení.
    • Může výrazně zlepšit výkon – latenci i propustnost. Proces serveru PostgreSQL musí vytvořit fork pro zpracování každého nového připojení a opětovným využitím připojení se této režii vyhnout.
    //DButil.java
package test.crud;

import java.io.FileInputStream;
import java.io.IOException;
import java.sql.SQLException;
import java.util.Properties;

import javax.sql.DataSource;

import com.zaxxer.hikari.HikariDataSource;

public class DButil {
    private static final String DB_USERNAME = "db.username";
    private static final String DB_PASSWORD = "db.password";
    private static final String DB_URL = "db.url";
    private static final String DB_DRIVER_CLASS = "driver.class.name";
    private static Properties properties =  null;
    private static HikariDataSource datasource;

    static {
        try {
            properties = new Properties();
            properties.load(new FileInputStream("src/main/java/application.properties"));

            datasource = new HikariDataSource();
            datasource.setDriverClassName(properties.getProperty(DB_DRIVER_CLASS ));
            datasource.setJdbcUrl(properties.getProperty(DB_URL));
            datasource.setUsername(properties.getProperty(DB_USERNAME));
            datasource.setPassword(properties.getProperty(DB_PASSWORD));
            datasource.setMinimumIdle(100);
            datasource.setMaximumPoolSize(1000000000);
            datasource.setAutoCommit(true);
            datasource.setLoginTimeout(3);
        } catch (IOException | SQLException  e) {
            e.printStackTrace();
        }
    }
    public static DataSource getDataSource() {
        return datasource;
    }
}

  2. V souboru src/main/java/vytvořte soubor DemoApplication.java , který obsahuje následující kód:

    package test.crud;
import java.io.IOException;
import java.sql.*;
import java.util.*;
import java.util.logging.Logger;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import org.postgresql.copy.CopyManager;
import org.postgresql.core.BaseConnection;
import java.io.IOException;
import java.io.Reader;
import java.io.StringReader;

public class DemoApplication {

    private static final Logger log;

    static {
        System.setProperty("java.util.logging.SimpleFormatter.format", "[%4$-7s] %5$s %n");
        log =Logger.getLogger(DemoApplication.class.getName());
    }
    public static void main(String[] args)throws Exception
    {
        log.info("Connecting to the database");
        Connection connection = DButil.getDataSource().getConnection();
        System.out.println("The Connection Object is of Class: " + connection.getClass());
        log.info("Database connection test: " + connection.getCatalog());
        log.info("Creating table");
        log.info("Creating index");
        log.info("distributing table");
        Scanner scanner = new Scanner(DemoApplication.class.getClassLoader().getResourceAsStream("schema.sql"));
        Statement statement = connection.createStatement();
        while (scanner.hasNextLine()) {
            statement.execute(scanner.nextLine());
        }
        log.info("Closing database connection");
        connection.close();
    }

}

    Poznámka:

    user Databáze a password přihlašovací údaje se používají při provádění DriverManager.getConnection(properties.getProperty("url"), properties);. Přihlašovací údaje jsou uloženy v souboru application.properties , který se předává jako argument.

  3. Teď můžete tuto hlavní třídu spustit pomocí svého oblíbeného nástroje:

    • Pomocí integrovaného vývojového DemoApplication prostředí (IDE) byste měli být schopni kliknout pravým tlačítkem myši na třídu a spustit ji.
    • Pomocí Mavenu můžete aplikaci spustit spuštěním následujícího příkazu:
      mvn exec:java -Dexec.mainClass="com.example.demo.DemoApplication".

Aplikace by se měla připojit ke službě Azure Cosmos DB for PostgreSQL, vytvořit schéma databáze a pak připojení zavřít, jak byste měli vidět v protokolech konzoly:

[INFO   ] Loading application properties
[INFO   ] Connecting to the database
[INFO   ] Database connection test: citus
[INFO   ] Create database schema
[INFO   ] Closing database connection

Vytvoření doménové třídy

Vytvořte novou Pharmacy třídu Java vedle DemoApplication třídy a přidejte následující kód:

public class Pharmacy {
    private Integer pharmacy_id;
    private String pharmacy_name;
    private String city;
    private String state;
    private Integer zip_code;
    public Pharmacy() { }
    public Pharmacy(Integer pharmacy_id, String pharmacy_name, String city,String state,Integer zip_code)
    {
        this.pharmacy_id = pharmacy_id;
        this.pharmacy_name = pharmacy_name;
        this.city = city;
        this.state = state;
        this.zip_code = zip_code;
    }

    public Integer getpharmacy_id() {
        return pharmacy_id;
    }

    public void setpharmacy_id(Integer pharmacy_id) {
        this.pharmacy_id = pharmacy_id;
    }

    public String getpharmacy_name() {
        return pharmacy_name;
    }

    public void setpharmacy_name(String pharmacy_name) {
        this.pharmacy_name = pharmacy_name;
    }

    public String getcity() {
        return city;
    }

    public void setcity(String city) {
        this.city = city;
    }

    public String getstate() {
        return state;
    }

    public void setstate(String state) {
        this.state = state;
    }

    public Integer getzip_code() {
        return zip_code;
    }

    public void setzip_code(Integer zip_code) {
        this.zip_code = zip_code;
    }
    @Override
    public String toString() {
        return "TPharmacy{" +
               "pharmacy_id=" + pharmacy_id +
               ", pharmacy_name='" + pharmacy_name + '\'' +
               ", city='" + city + '\'' +
               ", state='" + state + '\'' +
               ", zip_code='" + zip_code + '\'' +
               '}';
    }
}

Tato třída je doménový model mapovaný na Pharmacy tabulku, kterou jste vytvořili při provádění skriptu schema.sql .

Vložení dat

Do souboru DemoApplication.java za metodu main přidejte následující metodu, která pomocí příkazu INSERT INTO SQL vloží data do databáze:

private static void insertData(Pharmacy todo, Connection connection) throws SQLException {
    log.info("Insert data");
    PreparedStatement insertStatement = connection
        .prepareStatement("INSERT INTO pharmacy (pharmacy_id,pharmacy_name,city,state,zip_code)  VALUES (?, ?, ?, ?, ?);");

    insertStatement.setInt(1, todo.getpharmacy_id());
    insertStatement.setString(2, todo.getpharmacy_name());
    insertStatement.setString(3, todo.getcity());
    insertStatement.setString(4, todo.getstate());
    insertStatement.setInt(5, todo.getzip_code());

    insertStatement.executeUpdate();
}

Do hlavní metody přidejte dva následující řádky:

Pharmacy todo = new Pharmacy(0,"Target","Sunnyvale","California",94001);
insertData(todo, connection);

Spuštění hlavní třídy by teď mělo vytvořit následující výstup:

[INFO   ] Loading application properties
[INFO   ] Connecting to the database
[INFO   ] Database connection test: citus
[INFO   ] Creating table
[INFO   ] Creating index
[INFO   ] distributing table
[INFO   ] Insert data
[INFO   ] Closing database connection

Čtení dat

Přečtěte si data, která jste předtím vložili, a ověřte, že váš kód funguje správně.

Do souboru DemoApplication.java za metodu insertData přidejte následující metodu, která používá příkaz SELECT SQL ke čtení dat z databáze:

private static Pharmacy readData(Connection connection) throws SQLException {
    log.info("Read data");
    PreparedStatement readStatement = connection.prepareStatement("SELECT * FROM Pharmacy;");
    ResultSet resultSet = readStatement.executeQuery();
    if (!resultSet.next()) {
        log.info("There is no data in the database!");
        return null;
    }
    Pharmacy todo = new Pharmacy();
    todo.setpharmacy_id(resultSet.getInt("pharmacy_id"));
    todo.setpharmacy_name(resultSet.getString("pharmacy_name"));
    todo.setcity(resultSet.getString("city"));
    todo.setstate(resultSet.getString("state"));
    todo.setzip_code(resultSet.getInt("zip_code"));
    log.info("Data read from the database: " + todo.toString());
    return todo;
}

Do hlavní metody přidejte následující řádek:

todo = readData(connection);

Spuštění hlavní třídy by teď mělo vytvořit následující výstup:

[INFO   ] Loading application properties
[INFO   ] Connecting to the database
[INFO   ] Database connection test: citus
[INFO   ] Creating table
[INFO   ] Creating index
[INFO   ] distributing table
[INFO   ] Insert data
[INFO   ] Read data
[INFO   ] Data read from the database: Pharmacy{pharmacy_id=0, pharmacy_name='Target', city='Sunnyvale', state='California', zip_code='94001'}
[INFO   ] Closing database connection

Aktualizace dat

Aktualizujte data, která jste předtím vložili.

Stále v souboru DemoApplication.java za metodou readData přidejte následující metodu pro aktualizaci dat uvnitř databáze pomocí příkazu UPDATE SQL:

private static void updateData(Pharmacy todo, Connection connection) throws SQLException {
    log.info("Update data");
    PreparedStatement updateStatement = connection
        .prepareStatement("UPDATE pharmacy SET city = ? WHERE pharmacy_id = ?;");

    updateStatement.setString(1, todo.getcity());

    updateStatement.setInt(2, todo.getpharmacy_id());
    updateStatement.executeUpdate();
    readData(connection);
}

Do hlavní metody přidejte dva následující řádky:

todo.setcity("Guntur");
updateData(todo, connection);

Spuštění hlavní třídy by teď mělo vytvořit následující výstup:

[INFO   ] Loading application properties
[INFO   ] Connecting to the database
[INFO   ] Database connection test: citus
[INFO   ] Creating table
[INFO   ] Creating index
[INFO   ] distributing table
[INFO   ] Insert data
[INFO   ] Read data
[INFO   ] Data read from the database: Pharmacy{pharmacy_id=0, pharmacy_name='Target', city='Sunnyvale', state='California', zip_code='94001'}
[INFO   ] Update data
[INFO   ] Read data
[INFO   ] Data read from the database: Pharmacy{pharmacy_id=0, pharmacy_name='Target', city='Guntur', state='California', zip_code='94001'}
[INFO   ] Closing database connection

Odstranění dat

Nakonec odstraňte data, která jste předtím vložili. Stále v souboru DemoApplication.java za metodou updateData přidejte následující metodu pro odstranění dat uvnitř databáze pomocí příkazu DELETE SQL:

private static void deleteData(Pharmacy todo, Connection connection) throws SQLException {
    log.info("Delete data");
    PreparedStatement deleteStatement = connection.prepareStatement("DELETE FROM pharmacy WHERE pharmacy_id = ?;");
    deleteStatement.setLong(1, todo.getpharmacy_id());
    deleteStatement.executeUpdate();
    readData(connection);
}

Nyní můžete do hlavní metody přidat následující řádek:

deleteData(todo, connection);

Spuštění hlavní třídy by teď mělo vytvořit následující výstup:

[INFO   ] Loading application properties
[INFO   ] Connecting to the database
[INFO   ] Database connection test: citus
[INFO   ] Creating table
[INFO   ] Creating index
[INFO   ] distributing table
[INFO   ] Insert data
[INFO   ] Read data
[INFO   ] Data read from the database: Pharmacy{pharmacy_id=0, pharmacy_name='Target', city='Sunnyvale', state='California', zip_code='94001'}
[INFO   ] Update data
[INFO   ] Read data
[INFO   ] Data read from the database: Pharmacy{pharmacy_id=0, pharmacy_name='Target', city='Guntur', state='California', zip_code='94001'}
[INFO   ] Delete data
[INFO   ] Read data
[INFO   ] There is no data in the database!
[INFO   ] Closing database connection

Příkaz COPY pro rychlý příjem dat

Příkaz COPY může přinést obrovskou propustnost při příjmu dat do služby Azure Cosmos DB for PostgreSQL. Příkaz COPY může ingestovat data v souborech nebo z mikrodávek dat v paměti pro příjem dat v reálném čase.

Příkaz COPY pro načtení dat ze souboru

Následující kód zkopíruje data ze souboru CSV do databázové tabulky. Ukázka kódu vyžaduje soubor pharmacies.csv.

public static long
copyFromFile(Connection connection, String filePath, String tableName)
throws SQLException, IOException {
    long count = 0;
    FileInputStream fileInputStream = null;

    try {
        Connection unwrap = connection.unwrap(Connection.class);
        BaseConnection  connSec = (BaseConnection) unwrap;

        CopyManager copyManager = new CopyManager((BaseConnection) connSec);
        fileInputStream = new FileInputStream(filePath);
        count = copyManager.copyIn("COPY " + tableName + " FROM STDIN delimiter ',' csv", fileInputStream);
    } finally {
        if (fileInputStream != null) {
            try {
                fileInputStream.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    return count;
}

Nyní můžete do hlavní metody přidat následující řádek:

int c = (int) copyFromFile(connection,"C:\\Users\\pharmacies.csv", "pharmacy");
log.info("Copied "+ c +" rows using COPY command");

Spuštění main třídy by teď mělo vytvořit následující výstup:

[INFO   ] Loading application properties
[INFO   ] Connecting to the database
[INFO   ] Database connection test: citus
[INFO   ] Creating table
[INFO   ] Creating index
[INFO   ] distributing table
[INFO   ] Insert data
[INFO   ] Read data
[INFO   ] Data read from the database: Pharmacy{pharmacy_id=0, pharmacy_name='Target', city='Sunnyvale', state='California', zip_code='94001'}
[INFO   ] Update data
[INFO   ] Read data
[INFO   ] Data read from the database: Pharmacy{pharmacy_id=0, pharmacy_name='Target', city='Guntur', state='California', zip_code='94001'}
[INFO   ] Delete data
[INFO   ] Read data
[INFO   ] There is no data in the database!
[INFO ] Copied 5000 rows using COPY command
[INFO   ] Closing database connection

Příkaz COPY pro načtení dat v paměti

Následující kód zkopíruje data v paměti do tabulky.

private static void inMemory(Connection connection) throws SQLException,IOException
    {
    log.info("Copying inmemory data into table");
            
    final List<String> rows = new ArrayList<>();
    rows.add("0,Target,Sunnyvale,California,94001");
    rows.add("1,Apollo,Guntur,Andhra,94003");
        
    final BaseConnection baseConnection = (BaseConnection) connection.unwrap(Connection.class);
    final CopyManager copyManager = new CopyManager(baseConnection);

    // COPY command can change based on the format of rows. This COPY command is for above rows.
    final String copyCommand = "COPY pharmacy FROM STDIN with csv";        
       
    try (final Reader reader = new StringReader(String.join("\n", rows))) {
        copyManager.copyIn(copyCommand, reader);
    }
}

Nyní můžete do hlavní metody přidat následující řádek:

inMemory(connection);

Spuštění hlavní třídy by teď mělo vytvořit následující výstup:

[INFO   ] Loading application properties
[INFO   ] Connecting to the database
[INFO   ] Database connection test: citus
[INFO   ] Creating table
[INFO   ] Creating index
[INFO   ] distributing table
[INFO   ] Insert data
[INFO   ] Read data
[INFO   ] Data read from the database: Pharmacy{pharmacy_id=0, pharmacy_name='Target', city='Sunnyvale', state='California', zip_code='94001'}
[INFO   ] Update data
[INFO   ] Read data
[INFO   ] Data read from the database: Pharmacy{pharmacy_id=0, pharmacy_name='Target', city='Guntur', state='California', zip_code='94001'}
[INFO   ] Delete data
[INFO   ] Read data
[INFO   ] There is no data in the database!
5000
[INFO   ] Copying in-memory data into table
[INFO   ] Closing database connection

Opakování aplikace kvůli selhání žádostí o databázi

Někdy je možné, že databázové požadavky z vaší aplikace selžou. K takovým problémům může dojít v různých scénářích, jako je selhání sítě mezi aplikací a databází, nesprávné heslo atd. Některé problémy můžou být přechodné a během několika sekund až minut se vyřeší. Logiku opakování v aplikaci můžete nakonfigurovat tak, aby se překončily přechodné chyby.

Konfigurace logiky opakování ve vaší aplikaci pomáhá zlepšit uživatelské prostředí. Ve scénářích selhání uživatelé budou jen chvíli čekat, než aplikace bude obsluhovat požadavky, místo aby došlo k chybám.

Následující příklad ukazuje, jak implementovat logiku opakování ve vaší aplikaci. Ukázkový fragment kódu se pokusí o požadavek databáze každých 60 sekund (až pětkrát), dokud nebude úspěšný. Počet a četnost opakování je možné nakonfigurovat na základě potřeb vaší aplikace.

V tomto kódu nahraďte <cluster> názvem clusteru a <heslem> správce.

package test.crud;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.util.logging.Logger;
import com.zaxxer.hikari.HikariDataSource;

public class DemoApplication
{
    private static final Logger log;

    static
    {
        System.setProperty("java.util.logging.SimpleFormatter.format", "[%4$-7s] %5$s %n");
        log = Logger.getLogger(DemoApplication.class.getName());
    }
    private static final String DB_USERNAME = "citus";
    private static final String DB_PASSWORD = "<password>";
    private static final String DB_URL = "jdbc:postgresql://c-<cluster>.<uniqueID>.postgres.cosmos.azure.com:5432/citus?sslmode=require";
    private static final String DB_DRIVER_CLASS = "org.postgresql.Driver";
    private static HikariDataSource datasource;

    private static String executeRetry(String sql, int retryCount) throws InterruptedException
    {
        Connection con = null;
        PreparedStatement pst = null;
        ResultSet rs = null;
        for (int i = 1; i <= retryCount; i++)
        {
            try
            {
                datasource = new HikariDataSource();
                datasource.setDriverClassName(DB_DRIVER_CLASS);
                datasource.setJdbcUrl(DB_URL);
                datasource.setUsername(DB_USERNAME);
                datasource.setPassword(DB_PASSWORD);
                datasource.setMinimumIdle(10);
                datasource.setMaximumPoolSize(1000);
                datasource.setAutoCommit(true);
                datasource.setLoginTimeout(3);
                log.info("Connecting to the database");
                con = datasource.getConnection();
                log.info("Connection established");
                log.info("Read data");
                pst = con.prepareStatement(sql);
                rs = pst.executeQuery();
                StringBuilder builder = new StringBuilder();
                int columnCount = rs.getMetaData().getColumnCount();
                while (rs.next())
                {
                    for (int j = 0; j < columnCount;)
                    {
                        builder.append(rs.getString(j + 1));
                        if (++j < columnCount)
                            builder.append(",");
                    }
                    builder.append("\r\n");
                }
                return builder.toString();
            }
            catch (Exception e)
            {
                Thread.sleep(60000);
                System.out.println(e.getMessage());
            }
        }
        return null;
    }

    public static void main(String[] args) throws Exception
    {
        String result = executeRetry("select 1", 5);
        System.out.print(result);
    }
}

Další kroky