响应式 PostgreSQL 客户端（Reactive PostgreSQL Client）

<dependency>
  <groupId>io.vertx</groupId>
  <artifactId>vertx-pg-client</artifactId>
  <version>5.0.12</version>
</dependency>

dependencies {
  compile 'io.vertx:vertx-pg-client:5.0.12'
}

PgConnectOptions connectOptions = new PgConnectOptions()
  .setPort(5432)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret");

// 连接池选项
PoolOptions poolOptions = new PoolOptions()
  .setMaxSize(5);

// 创建客户端连接池
SqlClient client = PgBuilder
  .client()
  .with(poolOptions)
  .connectingTo(connectOptions)
  .build();

// 执行简单查询
client
  .query("SELECT * FROM users WHERE id='julien'")
  .execute()
  .onComplete(ar -> {
    if (ar.succeeded()) {
      RowSet<Row> result = ar.result();
      System.out.println("Got " + result.size() + " rows ");
    } else {
      System.out.println("Failure: " + ar.cause().getMessage());
    }

    // 关闭连接池
    client.close();
  });

PgConnectOptions connectOptions = new PgConnectOptions()
  .setPort(5432)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret");

// 连接池选项
PoolOptions poolOptions = new PoolOptions()
  .setMaxSize(5);

// 创建池化客户端
SqlClient client = PgBuilder
  .client()
  .with(poolOptions)
  .connectingTo(connectOptions)
  .build();

PgConnectOptions connectOptions = new PgConnectOptions()
  .setPort(5432)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret");

// 连接池选项
PoolOptions poolOptions = new PoolOptions()
  .setMaxSize(5);

// 创建池化客户端
SqlClient client = PgBuilder
  .client()
  .with(poolOptions)
  .connectingTo(connectOptions)
  .using(vertx)
  .build();

client.close();

PgConnectOptions connectOptions = new PgConnectOptions()
  .setPort(5432)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret");

// 连接池选项
PoolOptions poolOptions = new PoolOptions()
  .setMaxSize(5);

// 创建池化客户端
Pool pool = PgBuilder
  .pool()
  .with(poolOptions)
  .connectingTo(connectOptions)
  .using(vertx)
  .build();

// 从连接池获取连接
pool.getConnection().compose(conn -> {
  System.out.println("Got a connection from the pool");

  // 所有操作均在同一连接上执行
  return conn
    .query("SELECT * FROM users WHERE id='julien'")
    .execute()
    .compose(res -> conn
      .query("SELECT * FROM users WHERE id='emad'")
      .execute())
    .onComplete(ar -> {
      // 将连接释放回连接池
      conn.close();
    });
}).onComplete(ar -> {
  if (ar.succeeded()) {
    System.out.println("Done");
  } else {
    System.out.println("Something went wrong " + ar.cause().getMessage());
  }
});

Pool pool = PgBuilder
  .pool()
  .connectingTo(connectOptions.setPipeliningLimit(16))
  .with(poolOptions)
  .using(vertx)
  .build();

SqlClient client = PgBuilder
  .client()
  .with(poolOptions)
  .connectingTo(connectOptions)
  .using(vertx)
  .build();

// 流水线式执行
Future<RowSet<Row>> res1 = client.query(sql).execute();

// 连接池
Pool pool = PgBuilder
  .pool()
  .connectingTo(connectOptions)
  .with(poolOptions)
  .using(vertx)
  .build();

// 非流水线式执行
Future<RowSet<Row>> res2 = pool.query(sql).execute();

Pool pool = Pool.pool(database, new PoolOptions().setMaxSize(maxSize));
vertx.deployVerticle(() -> new VerticleBase() {
  @Override
  public Future<?> start() throws Exception {
    // 使用连接池
    return super.start();
  }
}, new DeploymentOptions().setInstances(4));

vertx.deployVerticle(() -> new VerticleBase() {
  Pool pool;
  @Override
  public Future<?> start() throws Exception {
    // 获取或创建共享连接池
    // 本质上是获取对该池的一份租约（Lease）
    // 当该 Verticle 卸载时，租约会自动释放
    pool = PgBuilder.pool()
      .with(new PoolOptions()
        .setMaxSize(maxSize)
        .setShared(true)
        .setName("my-pool"))
      .connectingTo(database)
      .using(vertx)
      .build();
    return super.start();
  }
}, new DeploymentOptions().setInstances(4));

Pool pool = PgBuilder.pool()
  .with(new PoolOptions()
    .setMaxSize(maxSize)
    .setShared(true)
    .setName("my-pool")
    .setEventLoopSize(4))
  .connectingTo(database)
  .using(vertx)
  .build();

PgConnectOptions connectOptions = new PgConnectOptions()
  .setHost("/var/run/postgresql")
  .setPort(5432)
  .setDatabase("the-db");

// 连接池选项
PoolOptions poolOptions = new PoolOptions()
  .setMaxSize(5);

// 创建池化客户端
Pool client = PgBuilder
  .pool()
  .connectingTo(connectOptions)
  .with(poolOptions)
  .using(vertx)
  .build();

options
  .setReconnectAttempts(2)
  .setReconnectInterval(1000);

client
  .query("SELECT * FROM users WHERE id='julien'")
  .execute()
  .onComplete(ar -> {
  if (ar.succeeded()) {
    RowSet<Row> result = ar.result();
    System.out.println("Got " + result.size() + " rows ");
  } else {
    System.out.println("Failure: " + ar.cause().getMessage());
  }
});

client
  .preparedQuery("INSERT INTO color (color_name) VALUES ($1), ($2), ($3) RETURNING color_id")
  .execute(Tuple.of("white", "red", "blue"))
  .onSuccess(rows -> {
    for (Row row : rows) {
      System.out.println("generated key: " + row.getInteger("color_id"));
    }
});

client
  .query("DELETE FROM color RETURNING color_name")
  .execute()
  .onSuccess(rows -> {
    for (Row row : rows) {
      System.out.println("deleted color: " + row.getString("color_name"));
    }
  });

client
  .preparedQuery("INSERT INTO color (color_name) VALUES ($1) RETURNING color_id")
  .executeBatch(Arrays.asList(Tuple.of("white"), Tuple.of("red"), Tuple.of("blue")))
  .onSuccess(res -> {
    for (RowSet<Row> rows = res; rows != null; rows = rows.next()) {
      Integer colorId = rows.iterator().next().getInteger("color_id");
      System.out.println("generated key: " + colorId);
    }
  });

connection
  .prepare("SELECT * FROM users WHERE first_name LIKE $1")
  .onComplete(ar0 -> {
  if (ar0.succeeded()) {
    PreparedStatement pq = ar0.result();

    // 游标要求在事务内部运行
    connection
      .begin()
      .onComplete(ar1 -> {
      if (ar1.succeeded()) {
        Transaction tx = ar1.result();

        // 创建游标
        Cursor cursor = pq.cursor(Tuple.of("julien"));

        // 读取 50 行
        cursor.read(50).onComplete(ar2 -> {
          if (ar2.succeeded()) {
            RowSet<Row> rows = ar2.result();

            // 检查是否还有剩余数据行？
            if (cursor.hasMore()) {
              // 重复读取流程...
            } else {
              // 全部数据行读取完毕 - 提交事务
              tx.commit();
            }
          }
        });
      }
    });
  }
});

cursor
  .read(50)
  .onComplete(ar2 -> {
  if (ar2.succeeded()) {
    // 关闭游标
    cursor.close();
  }
});

connection
  .prepare("SELECT * FROM users WHERE first_name LIKE $1")
  .onComplete(ar0 -> {
  if (ar0.succeeded()) {
    PreparedStatement ps = ar0.result();

    // 数据流要求在事务内部运行
    connection
      .begin()
      .onComplete(ar1 -> {
      if (ar1.succeeded()) {
        Transaction tx = ar1.result();

        // 每次抓取 50 行
        RowStream<Row> stream = pq.createStream(50, Tuple.of("julien"));

        // 使用数据流
        stream.exceptionHandler(err -> {
          System.out.println("Error: " + err.getMessage());
          ps.close();
        });
        stream.endHandler(v -> {
          // 关闭流以释放数据库中的游标资源
          stream
            .close()
            .onComplete(closed -> {
            tx.commit()
              .onComplete(committed -> {
                System.out.println("End of stream");
                ps.close();
            });
          });
        });
        stream.handler(row -> {
          System.out.println("User: " + row.getString("last_name"));
        });
      }
    });
  }
});

options.setTracingPolicy(TracingPolicy.ALWAYS);

pool
  .query("SELECT 1::BIGINT "VAL"")
  .execute()
  .onComplete(ar -> {
    RowSet<Row> rowSet = ar.result();
    Row row = rowSet.iterator().next();

    // 存储为 java.lang.Long
    Object value = row.getValue(0);

    // 转换为 java.lang.Integer
    Integer intValue = row.getInteger(0);
  });

pool
  .query("SELECT 1::BIGINT "VAL"")
  .execute()
  .onComplete(ar -> {
    RowSet<Row> rowSet = ar.result();
    Row row = rowSet.iterator().next();

    // 存储为 java.lang.Long
    Object value = row.getValue(0);

    // 转换为 java.lang.Integer
    Integer intValue = row.getInteger(0);
  });

Tuple tuple = Tuple.of(new String[]{ "a", "tuple", "with", "arrays" });

// 向元组添加字符串数组
tuple.addArrayOfString(new String[]{"another", "array"});

// 获取第一个字符串数组
String[] array = tuple.getArrayOfStrings(0);

client
  .query("SELECT 'infinity'::DATE "LocalDate"")
  .execute()
  .onComplete(ar -> {
    if (ar.succeeded()) {
      Row row = ar.result().iterator().next();
      System.out.println(row.getLocalDate("LocalDate").equals(LocalDate.MAX));
    } else {
      System.out.println("Failure: " + ar.cause().getMessage());
    }
  });

client
  .preparedQuery("SELECT address, (address).city FROM address_book WHERE id=$1")
  .execute(Tuple.of(3))
  .onComplete(ar -> {
  if (ar.succeeded()) {
    RowSet<Row> rows = ar.result();
    for (Row row : rows) {
      System.out.println("Full Address " + row.getString(0) + ", City " + row.getString(1));
    }
  } else {
    System.out.println("Failure: " + ar.cause().getMessage());
  }
});

client
  .preparedQuery("INSERT INTO address_book (id, address) VALUES ($1, $2)")
  .execute(Tuple.of(3, "('Anytown', 'Second Ave', false)"))
  .onComplete(ar -> {
  if (ar.succeeded()) {
    RowSet<Row> rows = ar.result();
    System.out.println(rows.rowCount());
  } else {
    System.out.println("Failure: " + ar.cause().getMessage());
  }
});

client
  .preparedQuery("SELECT to_tsvector( $1 ) @@ to_tsquery( $2 )")
  .execute(Tuple.of("fat cats ate fat rats", "fat & rat"))
  .onComplete(ar -> {
  if (ar.succeeded()) {
    RowSet<Row> rows = ar.result();
    for (Row row : rows) {
      System.out.println("Match : " + row.getBoolean(0));
    }
  } else {
    System.out.println("Failure: " + ar.cause().getMessage());
  }
});

client
  .preparedQuery("SELECT to_tsvector( $1 ), to_tsquery( $2 )")
  .execute(Tuple.of("fat cats ate fat rats", "fat & rat"))
  .onComplete(ar -> {
  if (ar.succeeded()) {
    RowSet<Row> rows = ar.result();
    for (Row row : rows) {
      System.out.println("Vector : " + row.getString(0) + ", query : "+row.getString(1));
    }
  } else {
    System.out.println("Failure: " + ar.cause().getMessage());
  }
});

client
  .preparedQuery("INSERT INTO colors VALUES ($2)")
  .execute(Tuple.of("red"))
  .onComplete(res -> {
    // ...
  });

client
  .preparedQuery("INSERT INTO colors VALUES ($1)")
  .execute(Tuple.of(Color.red))
  .flatMap(res ->
    client
      .preparedQuery("SELECT color FROM colors")
      .execute()
  ).onComplete(res -> {
    if (res.succeeded()) {
      RowSet<Row> rows = res.result();
      for (Row row : rows) {
        System.out.println(row.get(Color.class, "color"));
      }
    }
});

Collector<Row, ?, Map<Long, String>> collector = Collectors.toMap(
  row -> row.getLong("id"),
  row -> row.getString("last_name"));

// 结合收集器执行查询
client.query("SELECT * FROM users")
  .collecting(collector)
  .execute()
  .onComplete(ar -> {
  if (ar.succeeded()) {
    SqlResult<Map<Long, String>> result = ar.result();

    // 获取收集器生成的 Map
    Map<Long, String> map = result.value();
    System.out.println("Got " + map);
  } else {
    System.out.println("Failure: " + ar.cause().getMessage());
  }
});

Collector<Row, ?, String> collector = Collectors.mapping(
  row -> row.getString("last_name"),
  Collectors.joining(",", "(", ")")
);

// 结合收集器执行查询
client
  .query("SELECT * FROM users")
  .collecting(collector)
  .execute()
  .onComplete(ar -> {
    if (ar.succeeded()) {
      SqlResult<String> result = ar.result();

      // 获取收集器生成的字符串
      String list = result.value();
      System.out.println("Got " + list);
    } else {
      System.out.println("Failure: " + ar.cause().getMessage());
    }
  });

connection.notificationHandler(notification -> {
  System.out.println("Received " + notification.getPayload() + " on channel " + notification.getChannel());
});

connection
  .query("LISTEN some-channel")
  .execute()
  .onComplete(ar -> {
  System.out.println("Subscribed to channel");
});

PgSubscriber subscriber = PgSubscriber.subscriber(vertx, new PgConnectOptions()
  .setPort(5432)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret")
);

// 你可以在连接前设置订阅通道
subscriber.channel("channel1").handler(payload -> {
  System.out.println("Received " + payload);
});

subscriber
  .connect()
  .onComplete(ar -> {
  if (ar.succeeded()) {

    // 或者在连接成功后设置订阅通道
    subscriber.channel("channel2").handler(payload -> {
      System.out.println("Received " + payload);
    });
  }
});

PgSubscriber subscriber = PgSubscriber.subscriber(vertx, new PgConnectOptions()
  .setPort(5432)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret")
);

subscriber
  .connect()
  .onComplete(ar -> {
    if (ar.succeeded()) {
      // 复合通道名称 - 在 PostgreSQL 中需要用双引号标识符
      subscriber.channel("Complex.Channel.Name").handler(payload -> {
        System.out.println("Received " + payload);
      });
      subscriber.channel("Complex.Channel.Name").subscribeHandler(subscribed -> {
        subscriber.actualConnection()
          .query("NOTIFY "Complex.Channel.Name", 'msg'")
          .execute()
          .onComplete(notified -> {
            System.out.println("Notified "Complex.Channel.Name"");
          });
      });

      // PostgreSQL 简单标识符会被强制转换为小写
      subscriber.channel("simple_channel").handler(payload -> {
          System.out.println("Received " + payload);
      });
      subscriber.channel("simple_channel").subscribeHandler(subscribed -> {
        // 以下简单通道标识符会被强制转换为小写
        subscriber.actualConnection()
          .query("NOTIFY Simple_CHANNEL, 'msg'")
          .execute()
          .onComplete(notified -> {
            System.out.println("Notified simple_channel");
          });
      });

      // 以下通道名称超过了当前的限制
      // (NAMEDATALEN = 64) - 1 == 63 个字符，超出部分将被截断
      subscriber.channel("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaabbbbb")
        .handler(payload -> {
        System.out.println("Received " + payload);
      });
    }
  });

PgSubscriber subscriber = PgSubscriber.subscriber(vertx, new PgConnectOptions()
  .setPort(5432)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret")
);

// 最多重试 10 次，每次间隔 100 ms
subscriber.reconnectPolicy(retries -> {
  if (retries < 10) {
    return 100L;
  } else {
    return -1L;
  }
});

connection.noticeHandler(notice -> {
  System.out.println("Received notice " + notice.getSeverity() + "" + notice.getMessage());
});

connection
  .query("SELECT pg_sleep(20)")
  .execute()
  .onComplete(ar -> {
  if (ar.succeeded()) {
    // 假设这是一个长时间运行的查询并且仍在执行
    System.out.println("Query success");
  } else {
    // 服务器在收到取消请求后将中止当前查询
    System.out.println("Failed to query due to " + ar.cause().getMessage());
  }
});
connection
  .cancelRequest()
  .onComplete(ar -> {
  if (ar.succeeded()) {
    System.out.println("Cancelling request has been sent");
  } else {
    System.out.println("Failed to send cancelling request");
  }
});

PgConnectOptions options = new PgConnectOptions()
  .setPort(5432)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret")
  .setSslMode(SslMode.VERIFY_CA)
  .setSslOptions(new ClientSSLOptions().setTrustOptions(new PemTrustOptions().addCertPath("/path/to/cert.pem")));

PgConnection
  .connect(vertx, options)
  .onComplete(res -> {
  if (res.succeeded()) {
    // 已通过 SSL 成功连接
  } else {
    System.out.println("Could not connect " + res.cause());
  }
});

connectOptions.setUseLayer7Proxy(true);