Project Loom
This work will be simpler when Project Loom technology arrives in Java.
This project is adding virtual threads (fibers) to the Java concurrency toolbox. Many running virtual threads can be mapped to run on top of platform/kernel threads. When a virtual thread blocks, it is “parked”, and another virtual thread is assigned to execute on the platform/kernel thread. This switching between virtual threads is done very quickly, making thread-blocking extremely cheap in terms of its impact on performance.
Virtual threads are also extremely cheap in terms of its use of memory. Whereas platform/kernel threads are allocated rather large stack sizes no matter the need, virtual threads have a stack that expands as needed… and shrinks when no longer needed.
Virtual threads promise to eliminate the risks of using thread pools. Every thread is fresh with its own ThreadLocal values.
Experimental builds based on early-access Java 17 are available now. The Project Loom team is soliciting feedback.
AutoCloseable
In Loom, the ExecutorService interface becomes AutoCloseable. So we can use try-with-resources syntax. The flow-of-control leaves the try block only after all submitted tasks are done/failed/canceled. When leaving the try block, the executor service is automatically closed.
No need for CompletableFuture
You can simply launch many virtual threads, millions even, and let them run. Most of the need for the many methods on CompletableFuture evaporates. For more info, see the more recent presentations and interviews with Ron Pressler of Oracle.
We can simply spin off all the tasks on virtual threads, collecting Future objects along the way. Then simply wait for all those tasks to finish.
Example code
Establish an ExecutorService instance. Submit to that executor service your Callable tasks. Capture the returned Future objects to track successful completion.
int countTasks = 1_000 ; // Number of tasks to spin off into threads.
List < Future < YourResultClass > > futures = new ArrayList <>( countTasks );
try (
ExecutorService executorService = Executors.newVirtualThreadExecutor() ;
)
{
for ( int i = 0 ; i < countTasks ; i++ )
{
// Submit a Callable object to the executor service.
Future < YourResultClass > future = executorService.submit( ( ) -> {
// The work to be done in each thread.
YourResultClass yourResultObject = … ;
return yourResultObject;
} );
futures.add( future );
}
}
// At this point, flow-of-control blocks until all submitted tasks are done/failed/canceled.
// After this point, the executor service is automatically closed.
After the work is done, we can example the collected Future objects to verify results.
// Report on all the futures, all the results of the thousand tasks.
for ( Future < YourResultClass > future : futures )
{
try
{
System.out.println(
"future.isDone(): " + ( future.isDone() + " | future.isCompletedNormally(): " + future.isCompletedNormally() + " | future.isCancelled(): " + future.isCancelled() + " | result: " + future.get().toString() )
);
}
catch ( InterruptedException e )
{
e.printStackTrace();
}
catch ( ExecutionException e )
{
e.printStackTrace();
}
}
Example app
Here is a complete app. Not what I would do in production of course, but it makes for a decent demonstration I hope.
This code spins off a thousand tasks. Each task makes a REST call to ask Wikipedia for a random page. The contents of that page are then written to a H2 database. We collect the Future objects returned when submitting Callable tasks to the executor service, and we examine those after the work is done.
I configured an in-memory database, but you could just as well put the database in storage.
For simplicity, I defined my WikipediaPage class as a record. This new feature in Java 16 is a brief way to write a class whose main purpose is to immutable and transparently carry data. The compiler implicitly creates the constructor, getters, equals & hashCode, and toString. Not important to this Answer; you could just as well use a conventional class.
package work.basil.example.loopingfutures;
import org.h2.jdbcx.JdbcDataSource;
import javax.sql.DataSource;
import java.io.IOException;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.sql.*;
import java.time.*;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
public class App
{
public static void main ( String[] args )
{
System.out.println( Runtime.version() );
System.out.println( Runtime.getRuntime().maxMemory() );
App app = new App();
app.demo();
}
private void demo ( )
{
DataSource dataSource = this.configureDataSource();
this.destroyDatabaseContents( dataSource );
this.createDatabase( dataSource );
this.work( dataSource , 1_000 );
this.dumpTable( dataSource );
}
private DataSource configureDataSource ( )
{
JdbcDataSource dataSource = Objects.requireNonNull( new JdbcDataSource() ); // Implementation of `DataSource` bundled with H2.
dataSource.setURL( "jdbc:h2:mem:looping_loom_example_db;DB_CLOSE_DELAY=-1" ); // Set `DB_CLOSE_DELAY` to `-1` to keep in-memory database in existence after connection closes.
dataSource.setUser( "scott" );
dataSource.setPassword( "tiger" );
return dataSource;
}
private void destroyDatabaseContents ( DataSource dataSource )
{
try (
Connection conn = dataSource.getConnection() ;
)
{
String sql = """
DROP TABLE IF EXISTS wikipedia_page_
;
""";
System.out.println( "sql: \n" + sql );
try ( Statement stmt = conn.createStatement() ; )
{
stmt.execute( sql );
}
}
catch ( SQLException e )
{
e.printStackTrace();
}
}
private void createDatabase ( final DataSource dataSource )
{
try (
Connection conn = dataSource.getConnection() ;
)
{
String sql = """
CREATE TABLE wikipedia_page_
(
id_ UUID NOT NULL PRIMARY KEY ,
url_ VARCHAR NOT NULL ,
content_ CLOB NOT NULL ,
when_fetched_ TIMESTAMP WITH TIME ZONE NOT NULL ,
row_created_ TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP()
)
;
""";
System.out.println( "sql: \n" + sql );
try ( Statement stmt = conn.createStatement() ; )
{
stmt.execute( sql );
}
}
catch ( SQLException e )
{
e.printStackTrace();
}
}
private void work ( final DataSource dataSource , final int countPagesToFetchFromWikipedia )
{
List < Future < WikipediaPage > > futures = new ArrayList <>( countPagesToFetchFromWikipedia );
long start = System.nanoTime();
try (
ExecutorService executorService = newVirtualThreadExecutor() ;
)
{
for ( int i = 0 ; i < countPagesToFetchFromWikipedia ; i++ )
{
// Submit a Callable object to the executor service.
Future < WikipediaPage > future = executorService.submit( ( ) -> {
// To meet Wikipedia's limit of 200 requests per second, let's throttle by sleeping the worker thread.
try {Thread.sleep( Duration.ofMillis( 100 ) ); } catch ( InterruptedException e ) { e.printStackTrace(); }
WikipediaPage page = this.fetchPage();
this.persistPage( dataSource , page );
return page;
} );
futures.add( future );
}
}
// At this point, flow-of-control blocks until all submitted tasks are done/failed/canceled. The executor service is automatically closed.
Duration duration = Duration.ofNanos( System.nanoTime() - start );
System.out.println( "duration = " + duration + " for a count of " + countPagesToFetchFromWikipedia );
// Report on all the futures, all the results of the thousand tasks.
for ( Future < WikipediaPage > future : futures )
{
try
{
System.out.println(
"future.isDone(): " + ( future.isDone() + " | future.isCompletedNormally(): " + future.isCompletedNormally() + " | future.isCancelled(): " + future.isCancelled() + " | result: " + future.get().toString() )
);
}
catch ( InterruptedException e )
{
e.printStackTrace();
}
catch ( ExecutionException e )
{
e.printStackTrace();
}
}
System.out.println( "INFO - End of `work` method. Message # 26893b25-b09c-40d5-8cee-60e6a1d53852." );
}
private WikipediaPage fetchPage ( )
{
WikipediaPage page = null; // To be returned.
HttpClient client =
HttpClient
.newBuilder()
.followRedirects( HttpClient.Redirect.NORMAL )
.build();
HttpRequest request =
HttpRequest
.newBuilder()
.uri( URI.create( "https://en.wikipedia.org/api/rest_v1/page/random/summary" ) )
.build();
try
{
HttpResponse < String > response = client.send( request , HttpResponse.BodyHandlers.ofString() );
OffsetDateTime whenFetched = OffsetDateTime.now( ZoneOffset.UTC );
URI uri = response.uri();
String content = response.body();
System.out.println( "response = " + response );
System.out.println( "content = " + content );
page = new WikipediaPage( UUID.randomUUID() , uri.toString() , content , whenFetched );
}
catch ( IOException e )
{
e.printStackTrace();
}
catch ( InterruptedException e )
{
e.printStackTrace();
}
return Objects.requireNonNull( page );
}
private void persistPage ( final DataSource dataSource , final WikipediaPage wikipediaPage )
{
Objects.requireNonNull( dataSource );
Objects.requireNonNull( wikipediaPage );
String sql = """
INSERT INTO wikipedia_page_ ( id_ , url_ , content_ , when_fetched_ )
VALUES ( ? , ? , ? , ? )
;
""";
System.out.println( "sql: \n" + sql );
try (
Connection conn = dataSource.getConnection() ;
PreparedStatement pstmt = Objects.requireNonNull( conn ).prepareStatement( sql ) ;
)
{
pstmt.setObject( 1 , wikipediaPage.id );
pstmt.setString( 2 , wikipediaPage.url );
pstmt.setString( 3 , wikipediaPage.content );
pstmt.setObject( 4 , wikipediaPage.whenFetched );
int countRowsAffected = pstmt.executeUpdate();
if ( countRowsAffected != 1 )
{
System.out.println( "ERROR - Failed to insert row. Message # 4c3503d9-8cad-4e21-a625-c58054d9ca78." );
}
}
catch ( SQLException e )
{
e.printStackTrace();
}
}
private void dumpTable ( final DataSource dataSource )
{
String sql = "SELECT * FROM wikipedia_page_ ;";
System.out.println( "sql: \n" + sql );
try (
Connection conn = dataSource.getConnection() ;
Statement stmt = conn.createStatement() ;
ResultSet rs = stmt.executeQuery( sql ) ;
)
{
while ( rs.next() )
{
//Retrieve by column name
UUID id = rs.getObject( "id_" , UUID.class );
String url = rs.getString( "url_" );
String content = rs.getString( "content_" );
OffsetDateTime whenFetched = rs.getObject( "when_fetched_" , OffsetDateTime.class );
WikipediaPage wp = new WikipediaPage( id , url , content , whenFetched );
System.out.println( "wp = " + wp );
}
}
catch ( SQLException e )
{
e.printStackTrace();
}
}
record WikipediaPage(UUID id , String url , String content , OffsetDateTime whenFetched)
{
@Override
public String toString ( )
{
// Omitting the `content` field for brevity.
return "WikipediaPage{ " +
"id=" + id +
" | url='" + url + '\'' +
" | whenFetched=" + whenFetched +
" }";
}
}
}
Results
All of these results are for 1,000 tasks. Run on a Mac mini (2018) 3 GHz Intel Core i5 with 32 gigs of RAM, macOS Mojave 10.14.6, Java 17-loom+2-42 assigned maximum memory of 8 gigs (8589934592).
When running with Executors.newSingleThreadExecutor(), takes 7 minutes.
When running with Executors.newFixedThreadPool( 10 ), takes 1 minute.
When running with Executors.newFixedThreadPool( 100 ), takes half a minute.
When running with Executors.newVirtualThreadExecutor(), takes a quarter minute.
