kotlin半生对象

by osha1

由osha1

如何在Kotlin中使用Actor实现对象池 (How to implement an Object-Pool with an Actor in Kotlin)

We use object pool in jasync-sql to manage connections to the database. In this post, I will share how it is done in a performant, lock-free manner using Kotlin coroutines with an Actor.

我们在jasync-sql中使用对象池来管理与数据库的连接。在这篇文章中，我将分享如何通过与Actor一起使用Kotlin协程以高性能，无锁的方式完成操作。

An object pool has a very simple API to work with. It is a pool of objects with two methods: take() and return().

对象池具有非常简单的API。它是具有两种方法的对象池： take()和return() 。

On first sight it looks like a very simple problem. The main catch here is that it has to be both performant and thread-safe, and that’s what makes it interesting and tricky to implement.

乍看之下，这似乎是一个非常简单的问题。这里的主要问题是它必须兼具高性能和线程安全性，这就是实现它有趣且棘手的原因。

但是，嘿！为什么我们仍然需要一个对象池？ (But hey! Why do we need an object pool anyway?)

jasync-sql is a library to access relational databases like MySQL and PostgreSQL. Database connections are a great example of the need for object pools. The access to the database is done by obtaining a connection from a Connection-Pool, using it and returning it back to the pool.

jasync-sql是一个用于访问关系数据库(如MySQL和PostgreSQL)的库。数据库连接是需要对象池的一个很好的例子。通过从Connection-Pool获取连接，使用数据库并将其返回给池，可以完成对数据库的访问。

With a connection pool we get a couple of advantages over creating connections per each SQL query:

使用连接池，与每个SQL查询创建连接相比，我们有两个优点：

Reusing connections — since the overhead of initiating a connection to the database is high (handshake, etc), connection pools allow keeping connections alive, thus reducing that overhead.
重用连接 -由于启动与数据库的连接的开销很高(握手等)，因此连接池允许保持连接处于活动状态，从而减少了开销。
Limiting resources — creating a DB connection per user request can be overwhelming to the DB. Using a pool effectively adds a barrier, limiting the number of maximum number of concurrent connections.
限制资源 -为每个用户请求创建数据库连接可能会使数据库不堪重负。使用池有效地增加了障碍，限制了并发连接的最大数量。

Well, I am sold, but…

好吧，我被卖了，但是…

连接池在Java世界中解决不了吗？ (Isn’t a Connection Pool a solved problem in the Java world?)

Yes it is a solved problem if you’re using JDBC. In that case HikariCP is an excellent choice from my experience, but there are a lot of others. In the case of jasync-sql it is not possible to use HikariCP, because HikariCP works with the JDBC API, and the jasync-sql driver is not implementing that full-fledged API, only a subset of it.

是的，如果您使用JDBC，这是一个已解决的问题。在这种情况下，从我的经验来看， HikariCP是一个很好的选择，但还有很多其他选择。在jasync-sql的情况下，无法使用HikariCP ，因为HikariCP与JDBC API一起使用，并且jasync-sql驱动程序未实现该完整的API，仅实现了一部分。

What about other Object pools in Java world?

Java世界中的其他对象池又如何呢？

There are numerous implementations, but it turns out that you usually find some specific requirement that was not implemented by that pool you’re using.

有许多实现，但是事实证明，您通常会发现一些特定要求，而该要求不是您所使用的池所实现的。

In our case, that requirement was non-blocking. In our pool, all operations have to be non-blocking since the library is async. For example, the take() operation in most implementations returns an object immediately or blocks until an object is ready. Our take() returns a Future<Connection>, which will be completed and continued when the connection is ready to use.

在我们的情况下，该要求是无障碍的。在我们的池中，由于库是异步的，因此所有操作都必须是非阻塞的。例如，大多数实现中的take()操作立即返回一个对象或阻塞直到对象准备就绪。我们的take()返回Future<Connecti on>，当连接准备就绪时，它将完成并继续。

I haven’t seen such an implementation in the wild.

我还没有在野外看到这样的实现。

I really like this answer from Stack Exchange:

我真的很喜欢来自Stack Exchange的答案：

Is object pooling a deprecated technique?Software Engineering Stack Exchange is a question and answer site for professionals, academics, and students working…softwareengineering.stackexchange.com

对象池是否已被弃用？ 软件工程堆栈交换是一个为专业人士，学者和工作的学生提供的问答网站 。softwareengineering.stackexchange.com

Another requirement that makes it hard to find an alternative is the need to try and stay compatible as much as possible with the current implementation we have.

另一个很难找到替代方法的要求是，需要尝试与现有的实现尽可能保持兼容。

In case you want to see other implementations you can check here:

如果您想查看其他实现，可以在这里查看：

object pool in java - Google Searchobject pool is a collection of a particular object that an application will create and keep on hand for those…www.google.co.il

Java中的对象池-Google搜索 对象池是应用程序将创建的特定对象的集合，并将这些对象保存在手中…… www.google.co.il

那么我们如何实现对象池呢？ (So how did we implement Object Pool?)

Before we dive into the details, let’s observe other requirements from the object pool that were omitted above for clarity but are necessary details.

在深入研究细节之前，让我们观察一下对象池中的其他需求，为清晰起见，这些需求在上面已被省略，但它们是必要的细节。

介面 (Interfaces)

The Object pool interface looks like this:

对象池接口如下所示：

interface AsyncObjectPool<T&gt; {  fun take(): CompletableFuture&lt;T>  fun giveBack(item: T): CompletableFuture<AsyncObjectPool<T>>  fun close(): CompletableFuture<AsyncObjectPool<T>>

In addition, when a pool wants to create new objects (connections) it will call the ObjectFactory. The factory has a couple more methods to handle the object lifecycle:

另外，当池要创建新对象(连接)时，它将调用ObjectFactory 。工厂有更多其他方法来处理对象生命周期：

validate — a method to check that the object is still valid. The method should be fast and check only in-memory constructs. For connections we usually check that the last query did not throw an exception and did not get a termination message from netty.
validate — 验证对象仍然有效的方法。该方法应该是快速的，并且仅检查内存中的构造。对于连接，我们通常检查最后一个查询没有引发异常，也没有从netty获得终止消息。
test — similar to validate, but a more exhaustive check. We allow test method to be slow and access the network etc. This method is used to check that idle objects are still valid. For connections, that will be something similar to select 0.
测试 -与验证类似，但检查更为详尽。我们允许测试方法变慢并访问网络等。此方法用于检查空闲对象是否仍然有效。对于连接，这类似于select 0 。
destroy — called to clean up the object when the pool is not using it anymore.
销毁 -在池不再使用该对象时调用以清除该对象。

The complete interface is:

完整的界面是：

interface ObjectFactory<T> {  fun create(): CompletableFuture<;out T>  fun destroy(item: T)  fun validate(item: T): Try<T>  fun test(item: T): CompletableFuture<T>

For pool configuration we have the following properties:

对于池配置，我们具有以下属性：

maxObjects — maximum number of connections we allow.
maxObjects —我们允许的最大连接数。
maxIdle — time that we leave the connection open without use. After that time it will be reclaimed.
maxIdle —我们不使用连接而保持打开状态的时间。在那之后，它将被回收。
maxQueueSize — when a request for a connection arrives and no connection is available, we put the request on hold in a queue. In case the queue is full (its size passed maxQueueSize) it will not wait but instead return an error.
maxQueueSize —当一个连接请求到达并且没有可用的连接时，我们将该请求置于队列中。如果队列已满(其大小通过maxQueueSize传递)，它将不等待而是返回一个错误。
createTimeout — maximum time to wait for a new connection to be created.
createTimeout —等待创建新连接的最长时间。
testTimeout — maximum time to wait for a test query on an idle connection. If it passes we will consider the connection as erroneous.
testTimeout —在空闲连接上等待测试查询的最长时间。如果通过，我们将认为连接错误。
validationInterval — on this interval, we will test if the idle connections are active and free up connections that passed maxIdle. We will also remove connections that passed testTimeout.
validationInterval -在此期间，我们将测试如果空闲连接有效且腾出通过连接maxIdle 。我们还将删除通过testTimeout连接。

原始实施 (Original implementation)

The first implementation of object pool was single threaded. All operations were sent to a worker thread that was responsible to execute them. This method is known as thread-confinement. Object creation and test operations were blocking and query execution itself was non-blocking.

对象池的第一个实现是单线程的。所有操作都发送到负责执行这些操作的工作线程。此方法称为线程约束。对象创建和测试操作处于阻塞状态，而查询执行本身是非阻塞的。

This method is problematic because operations are done one after another. On top of that, there are a couple of operations that are blocking as mentioned above. There were various cases of high latency when working in some scenarios and use cases (like here for example).

这种方法有问题，因为操作是一个接一个地完成的。最重要的是，如上所述，有几项操作正在阻塞。在某些情况和用例中工作时，存在各种高延迟情况(例如，例如此处 )。

As a workaround PartitionedPool was introduced. This is a workaround to the block issue with the above single-threaded approach. The partitioned pool creates multiple SingleThreadedObjectPools, each with its own worker. When a connection is requested, a pool is selected by a modulus on the thread id. The partitioned pool is actually a pool of pools ;-)

作为解决方法，引入了PartitionedPool 。这是上述单线程方法解决块问题的一种解决方法。分区池创建多个SingleThreadedObjectPools ，每个都有其自己的工作程序。当请求连接时，将通过线程ID上的模数来选择一个池。分区池实际上是池的池;-)

I mentioned this is a workaround since it has its own problems: you might still be blocking, but at a lower rate — plus it consume more threads and resources.

我提到了这是一种解决方法，因为它有其自身的问题：您可能仍在阻塞，但是速率较低-而且它消耗更多的线程和资源。

基于Actor的实现 (Actor based implementation)

An Actor is an entity that has a mailbox. It receives messages to its mailbox and processes them one after the other. The mailbox is a sort of a channel to pass events from the outside world to the actor.

Actor是具有邮箱的实体。它接收到其邮箱的消息，然后一个接一个地处理它们。邮箱是一种将事件从外界传递给演员的渠道。

A coroutines actor employs lock-free algorithms to allow fast and performant execution of events without the need for locks and synchronized blocks.

协程演员使用无锁算法来快速，高效地执行事件，而无需锁和synchronized块。

You can see an elaborated explanation here.

您可以在此处看到详细的说明。

In our case those events will be take and giveBack. In addition to those, we will have internal messages that the actor sends to itself like objectCreated etc. That allows the actor to have states that does not suffer from concurrency problems, as it is always confined to the same sequential execution. In addition the channel that passes those events is a queue that is using lock-free algorithms so it is very efficient, avoids contention, and generally has very high performance.

在我们的情况下，这些事件将为take和giveBack 。除此之外，我们还将拥有objectCreated发送给自己的内部消息，例如objectCreated等。这使得objectCreated具有不受并发问题困扰的状态，因为它始终限于同一顺序执行。另外，传递这些事件的通道是使用无锁算法的队列，因此它非常高效，避免争用并且通常具有很高的性能。

There is an excellent video explaining how this was implemented (note that this is “heavy” algorithmic staff):

有一个精彩的视频解释了如何实现(请注意，这是“繁重的”算法工作人员)：

Let’s recap what we have until now:

让我们回顾一下到目前为止所拥有的：

An actor receives messages and processes them one by one.
演员接收消息并对其进行逐一处理。
Usually messages will contain a CompletableFuture that should be completed when the actor processes it.
通常，消息将包含CompletableFuture ，当actor处理该消息时应将其完成。

Messages will be completed immediately or delayed (like in case we are waiting for a connection to be created). If it is delayed the actor will put the Future in a queue, and will use a callback mechanism to notify itself when the original future can be completed.

消息将立即完成或延迟(例如，在我们等待连接建立的情况下)。如果延迟，则参与者将把Future放在队列中，并将使用回调机制通知自己何时可以完成原始的Future 。

Message processing in the actor should not be blocked or delay the actor. If this happens, it will delay all messages waiting to be processed in the queue and will slow down the entire actor operation.
actor中的消息处理不应被阻塞或延迟actor。如果发生这种情况，它将延迟所有等待在队列中处理的消息，并且会减慢整个actor操作的速度。

That’s why, in case we have long running operations inside the actor, we use the callback mechanism.

这就是为什么如果我们在actor中长时间运行操作的原因，我们使用回调机制。

让我们看一下用例的更多细节 (Let’s see more details on the use cases)

Take — someone wants an object from the pool. It will send a message with a callback to the actor. The actor will do one of the following things:

Take -有人要从游泳池里Take东西。它将带有回调的消息发送给参与者。演员将执行以下操作之一：

If the object is available — the actor will simply return it.
如果对象可用-演员将简单地将其返回。
If the pool hasn’t passed the limit of created objects — the actor will create a new object and return it when the object is ready.
如果池尚未通过创建对象的限制，则actor将创建一个新对象，并在对象准备就绪时将其返回。

In such a case, object creation can take time, so the actor will connect the callback from the object creation to the original take request callback.

在这种情况下，对象创建可能会花费一些时间，因此参与者会将回调从对象创建连接到原始的获取请求回调。

Will put the request in a queue for an available object (unless the queue is full and in that case will just return an error).
将请求放入可用对象的队列中(除非队列已满，在这种情况下只会返回错误)。

GiveBack — someone wants to give an object back to the pool (release it). This is also done by a message to the actor. The actor will do one of the following:

GiveBack有人想将对象还给池(释放它)。这也可以通过发送给演员的消息来完成。演员将执行以下操作之一：

If someone is waiting on the wait queue — it will borrow the object to it.
如果有人在等待队列中等待，它将向其借用该对象。
In other cases it will just keep the object on the pool for requests to come, so the object remains idle.
在其他情况下，它将对象仅保留在池中以等待请求，因此该对象保持空闲状态。

Test — periodically, someone from outside will notify the actor to test connections:

Test -定期，外部人员会通知参与者测试连接：

The actor will release the idle connection that hasn’t been used for a long time (it’s configurable).
actor将释放很长时间未使用的空闲连接(它是可配置的)。
The actor will test other idle objects using the ObjectFactory. It will send a callback to the factory and mark those objects as In Use, to prevent from borrowing them until the test is completed.
参与者将使用ObjectFactory测试其他空闲对象。它将向工厂发送一个回调并将这些对象标记为In Use ，以防止在测试完成之前借用它们。
The actor will check for timeouts in tests and destroy time-outed objects.
参与者将检查测试中的超时并破坏超时的对象。

Those are the main use cases.

这些是主要的用例。

泄漏 (Leaks)

There can be all sort of leaks in an object pool. Some are internal bugs which I hope are easier to spot and fix, and others are objects that were taken but not returned due to some user error. In such cases, objects might remain in the “In Use” queue forever.

对象池中可能存在各种泄漏。我希望一些内部错误更容易发现和修复，而其他一些则是由于某些用户错误而被拿回但未返回的对象。在这种情况下，对象可能永远保留在“ 使用中”队列中。

To avoid such cases, the “In Use” Map is using Java’s WeakHashMap. So if a user lost a connection it will be automatically removed from the map when it is cleaned by Java’s Garbage-Collector.

为了避免这种情况， “使用中”地图使用Java的WeakHashMap 。因此，如果用户失去了连接，当Java的Garbage-Collector清理连接时，它将自动从地图中删除。

In addition we added a log message in such cases that says: “LEAK-DETECTED”.

此外，在这种情况下，我们添加了一条日志消息，内容为： “泄漏检测到” 。

而已！ (That’s it!)

The full Kotlin source code of the object pool is available here:

对象池的完整Kotlin源代码在此处提供：

jasync-sql/jasync-sqlJava async database driver for MySQL and PostgreSQL written in Kotlin - jasync-sql/jasync-sqlgithub.com

jasync-sql / jasync-sql 用Kotlin编写的用于MySQL和PostgreSQLJava异步数据库驱动程序-jasync-sql / jasync-sql github.com

In an upcoming post I will compare performance metrics of the different implementations.

在下一篇文章中，我将比较不同实现的性能指标。

If you want to read more about Kotlin there is a nice introduction here:

如果您想了解有关Kotlin的更多信息，这里有一个不错的介绍：

And for coroutines in general check out this video:

对于一般的协程，请观看以下视频：

Finally if you want to learn more about Actors implementation using coroutines in Kotlin, then head over here:

最后，如果您想了解有关Kotlin中使用协程的Actor实现的更多信息，请前往此处：

Kotlin/kotlinx.coroutinesLibrary support for Kotlin coroutines . Contribute to Kotlin/kotlinx.coroutines development by creating an account on…github.com

Kotlin / kotlinx.coroutines库特林 协同程序的库支持。 通过在 github.com 上创建一个帐户，为Kotlin / kotlinx.coroutines开发做出贡献

Thanks for reading! ❤️

谢谢阅读！ ❤️