Corda RPC 连接池/缓存

Question

Corda有连接池功能吗？如何处理多个 RPC 用户连接池...

如果您可以重定向到 RPC 连接池/缓存的任何开源实现/指南，请不胜感激...

Answer 1

下面是如何池化节点 RPC 连接的示例：

import net.corda.client.rpc.CordaRPCClient
import net.corda.client.rpc.CordaRPCConnection
import net.corda.core.utilities.NetworkHostAndPort
import net.corda.core.utilities.contextLogger
import net.corda.core.utilities.getOrThrow
import net.corda.node.services.Permissions
import net.corda.testing.driver.DriverParameters
import net.corda.testing.driver.NodeParameters
import net.corda.testing.driver.driver
import net.corda.testing.node.User
import org.junit.Test
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.LinkedBlockingQueue
data class UserParams(val username: String, val password: String)
class PooledRpcConnections(address: NetworkHostAndPort): AutoCloseable {
val userToPool = ConcurrentHashMap<UserParams, LinkedBlockingQueue<CordaRPCConnection>>()
val client = CordaRPCClient(address)
fun <A> withConnection(userParams: UserParams, block: (CordaRPCConnection) -> A): A {
val queue = userToPool.getOrPut(userParams) { LinkedBlockingQueue() }
val connection = queue.poll() ?: client.start(userParams.username, userParams.password)
return try {
block(connection)
} finally {
queue.add(connection)
}
}
override fun close() {
for (queue in userToPool.values) {
do {
val connection = queue.poll()
connection?.close()
} while (connection != null)
}
}
}
class Test {
companion object {
val log = contextLogger()
}
@Test
fun poolWorks() {
val users = ('a' .. 'f').map { User(it.toString(), it.toString(), setOf(Permissions.all())) }
val userParams = users.map { UserParams(it.username, it.password) }
driver(DriverParameters(startNodesInProcess = true, notarySpecs = emptyList())) {
log.info("Starting node for users ${users.map { it.username }}")
val node = startNode(NodeParameters(rpcUsers = users)).getOrThrow()
log.info("Starting pool")
PooledRpcConnections(node.rpcAddress).use { pool ->
val N = 1000
log.info("Making $N requests using pooled connections")
(1 .. N).toList().parallelStream().forEach { i ->
val user = userParams[i % users.size]
pool.withConnection(user) { connection ->
log.info("USER[${user.username}] CONNECTION[${connection.hashCode()}] NODE_TIME[${connection.proxy.currentNodeTime()}]")
}
}
log.info("Done! Number of connections used per user: ${pool.userToPool.map { it.key.username to it.value.size }}")
}
}
}
}