toArray(): Returns an array containing the elements of the stream.
List<Student> listOfStudents = ...
Student[] arrayOfStudents = students().toArray(Student[]::new);List<Integer> numbers = ...
Integer[] numberArray = numbers.stream().toArray(Integer[]::new);
System.out.println(Arrays.toString(numberArray)); // prints [1, 2, 3, 4, 5]skip(): Returns a stream with the specified number of elements skipped.
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
List<Integer> skippedNumbers = numbers.stream().skip(3).collect(Collectors.toList());
System.out.println(skippedNumbers); // prints [4, 5]List<String> tokens = ...;
tokens.stream().skip(1).forEach(token -> ...);
//noinspection resourceInputStream inputStream = null;try {inputStream = new FileInputStream("file.txt");// do something with inputStream} catch (IOException e) {// handle exception} finally {// close inputStream}
import java.lang.invoke.VarHandle;
As a convention, we should declare VarHandles as static final fields and explicitly initialize them in static blocks. Also, we usually use the uppercase version of the corresponding field name as their name.
public class VariableHandlesUnitTest {
public int publicTestVariable = 1;
private int privateTestVariable = 1;
public int variableToSet = 1;
public int variableToCompareAndSet = 1;
public int variableToGetAndAdd = 0;
public byte variableToBitwiseOr = 0;
VarHandle PUBLIC_TEST_VARIABLE = MethodHandles
.lookup()
.in(VariableHandlesUnitTest.class) // Class name
.findVarHandle(VariableHandlesUnitTest.class, // Class name
"publicTestVariable", // Variable name
int.class // Variable type
);
VARIABLE_TO_COMPARE_AND_SET.compareAndSet(this, 1, 100); assertEquals(100, (int) VARIABLE_TO_COMPARE_AND_SET.get(this));
assertEquals(1, PUBLIC_TEST_VARIABLE.coordinateTypes().size()); assertEquals(VariableHandlesUnitTest.class,PUBLIC_TEST_VARIABLE.coordinateTypes().get(0));
int before = (int) VARIABLE_TO_GET_AND_ADD.getAndAdd(this, 200); assertEquals(0, before); assertEquals(200, (int) VARIABLE_TO_GET_AND_ADD.get(this));
byte before = (byte) VARIABLE_TO_BITWISE_OR.getAndBitwiseOr(this, (byte) 127); assertEquals(0, before); assertEquals(127, (byte) VARIABLE_TO_BITWISE_OR.get(this));
Supplier<CompletionStage<Void>> task = queue.poll();
try {
task.get()
.whenCompleteAsync((value, exception) -> {
if (throwable != null) {
logger.warning("...", throwable);
})
.exceptionally(exception -> ...);
} catch (Throwable exception) {
...
}
// Bir değer nesnesirecord FieldValue(long value){}// Bir işlem tipirecord Aggregator(long id, Type type)public enum Type {SUM, AVG, MAX, MIN;}}// Bir işlem ve ona uygulanacak değerMap<Aggregator, FieldValue> values = Map.of(new Aggregator(1L, Type.AVG), new FieldValue(10L),new Aggregator(2L, Type.AVG), new FieldValue(20L),new Aggregator(3L, Type.SUM), new FieldValue(30L),new Aggregator(4L, Type.SUM), new FieldValue(40L),new Aggregator(5L, Type.MAX), new FieldValue(50L),new Aggregator(6L, Type.MAX), new FieldValue(60L),new Aggregator(7L, Type.MIN), new FieldValue(70L),new Aggregator(8L, Type.MIN), new FieldValue(80L));
Map<Aggregator.Type, Double> result = aggregate(values);
public Map<AggregatorType, Double> aggregate(Map<A, B> fields) {
// Aynı işlem tipine göre değerleri grupla
Map<AggregatorType, List<Long>> valuesByType = fields.entrySet()
.stream()
.collect(Collectors.groupingBy(
entry -> entry.getKey().type(),
Collectors.mapping(
entry -> entry.getValue().value(),
Collectors.toList())
));
// Her işlem tipine göre listeyi hesapla
return valuesByType.entrySet()
.stream()
.collect(Collectors.toMap(
entry -> entry.getKey(),
entry -> collectValues(entry.getKey(), entry.getValue())
));
}
private double collectValues(AggregatorType aggregator, List<Long> values) {
DoubleStream stream = values.stream().mapToDouble(x -> x + 0.0d);
return switch (aggregator) {
case AVG -> stream.average().orElseThrow();
case SUM -> stream.sum();
case MAX -> stream.max().orElseThrow();
case MIN -> stream.min().orElseThrow();
default -> throw new IllegalArgumentException();
};
}@RequiredArgsConstructor
public enum Type {
SUM(values -> doubleStream(values).sum()),
AVG(values -> doubleStream(values).average().orElseThrow()),
MAX(values -> doubleStream(values).max().orElseThrow()),
MIN(values -> doubleStream(values).min().orElseThrow());
private final Function<List<Long>, Double> aggregator;
public Double aggregate(List<Long> values) {
return aggregator.apply(values);
}
private static DoubleStream doubleStream(List<Long> values) {
return values.stream().mapToDouble(v -> v + 0.0d);
}
}public Map<Aggregator.Type, Double> aggregate(Map<Aggregator, FieldValue> fields) {
// Aynı işlem tipine göre değerleri gruplaa
Map<Aggregator.Type, List<Long>> valuesByType = ...
return valuesByType.entrySet()
.stream()
.collect(Collectors.toMap(
entry.getKey(),
entry.getKey().aggregate(entry.getValue()) // Switch yerine Type nesnesini kullan
));
}import com.zaxxer.hikari.util.DriverDataSource;
@Bean public DataSource dataSource() { return new DriverDataSource("jdbc:mysql://127.0.0.1:5701", "com.hazelcast.datastore.JdbcDataStoreFactory", new Properties(), "", ""); }
@Getter @Setter @NoArgsConstructor public class Car { private String manufacturer; private String model; private String engineType; private Integer power; }
```javaMy code```
```xmlMy code```
One of the things that are unique about Micrometer is that it aims to be a facade with support for multiple observability standards, including OTEL, Brave, and others.As such, its focus is more on defining the vocabulary, conventions, and interfaces to use in the application which are arguably the more important aspect of making logging effective from a code perspective.
@Bean
public OpenTelemetry getOpenTelemetry() {
Resource serviceNameResource = Resource.create(
Attributes.of(ResourceAttributes.SERVICE_NAME, applicationName));
//[OTel component] SpanExporter is a component that gets called when a span is finished.
OtlpGrpcSpanExporter otlpExporter =
OtlpGrpcSpanExporter.builder()
.setEndpoint(otlpEexporterEndpoint)
.setTimeout(30, TimeUnit.SECONDS)
.build();
// [OTel component] SdkTracerProvider is an SDK implementation for TracerProvider
SdkTracerProvider sdkTracerProvider =
SdkTracerProvider.builder()
.setResource(Resource.getDefault().merge(serviceNameResource))
.addSpanProcessor(SimpleSpanProcessor.create(otlpExporter))
.build();
// [OTel component] The SDK implementation of OpenTelemetry
OpenTelemetrySdk openTelemetrySdk =
OpenTelemetrySdk.builder()
.setTracerProvider(sdkTracerProvider)
.setPropagators(ContextPropagators.create(B3Propagator.injectingSingleHeader()))
.build();
return openTelemetrySdk;
}@PostMapping("/pets/new")public String processCreationForm(Owner owner, @Valid Pet pet, BindingResult result, ModelMap model) { Observation observation = Observation.createNotStarted("create_pet", registry).start(); try (Observation.Scope scope = observation.openScope()) { // only 2 names are valid (low cardinality) observation.lowCardinalityKeyValue("pet.type", pet.getType().getName()); observation.highCardinalityKeyValue("pet.name", pet.getName()); observation.event(Observation.Event.of("PetCreated")); //more logic this.owners.save(owner); return "redirect:/owners/{ownerId}"; } catch (Exception exception) { observation.error(exception); throw exception; } finally { observation.stop(); } }
The method observe is actually a timer that will measure the execution time of whatever is passed inside.
@PostMapping("/pets/{petId}/edit")public String processUpdateForm(@Valid Pet pet, BindingResult result, Owner owner, ModelMap model) { return Observation.createNotStarted("update ", registry).observe(()->{ if (result.hasErrors()) { model.put("pet", pet); return VIEWS_PETS_CREATE_OR_UPDATE_FORM; } owner.addPet(pet); this.owners.save(owner); return "redirect:/owners/{ownerId}"; }); }
public class UserController {private final ObservationRegistry observationRegistry;@GetMappingpublic User findUserByFirstName(@RequestParam("firstName") String firstName) {return Observation.createNotStarted("users.find", this.observationRegistry).observe(() -> {Optional<User> findUserResult = this.userRepository.findByFirstName(firstName);if (findUserResult.isEmpty()) {throw new IllegalArgumentException( String.format("User with firstName=%s not found.", firstName));}return findUserResult.get();});}}
private static void virtualThreadsDemo(int numberOfThreads) {try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {IntStream.range(0, numberOfThreads).forEach(i -> {executor.submit(() -> {Thread.sleep(Duration.ofMillis(1));return i;});});}}
try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) { for (int i = 1; i <= 10; i++) { String taskName = "Task" + i; executor.execute(() -> someWork(taskName)); } }
VirtualThread[#25]/runnable@ForkJoinPool-1-worker-4 executing Task4VirtualThread[#26]/runnable@ForkJoinPool-1-worker-5 executing Task5VirtualThread[#28]/runnable@ForkJoinPool-1-worker-7 executing Task7VirtualThread[#23]/runnable@ForkJoinPool-1-worker-2 executing Task2VirtualThread[#21]/runnable@ForkJoinPool-1-worker-1 executing Task1VirtualThread[#24]/runnable@ForkJoinPool-1-worker-3 executing Task3VirtualThread[#27]/runnable@ForkJoinPool-1-worker-6 executing Task6VirtualThread[#29]/runnable@ForkJoinPool-1-worker-8 executing Task8VirtualThread[#32]/runnable@ForkJoinPool-1-worker-3 executing Task10VirtualThread[#31]/runnable@ForkJoinPool-1-worker-8 executing Task9VirtualThread[#26]/runnable@ForkJoinPool-1-worker-1 completed Task5VirtualThread[#21]/runnable@ForkJoinPool-1-worker-1 completed Task1VirtualThread[#25]/runnable@ForkJoinPool-1-worker-9 completed Task4VirtualThread[#24]/runnable@ForkJoinPool-1-worker-2 completed Task3VirtualThread[#27]/runnable@ForkJoinPool-1-worker-2 completed Task6VirtualThread[#32]/runnable@ForkJoinPool-1-worker-9 completed Task10VirtualThread[#28]/runnable@ForkJoinPool-1-worker-1 completed Task7VirtualThread[#23]/runnable@ForkJoinPool-1-worker-7 completed Task2VirtualThread[#31]/runnable@ForkJoinPool-1-worker-6 completed Task9VirtualThread[#29]/runnable@ForkJoinPool-1-worker-1 completed Task8
JDK 19 introduces structured concurrency, which is a multi-threaded programming method designed to simplify multi-threaded programming through the structured concurrency API, not to replace java.util.concurrent.Structured concurrency treats multiple tasks running in different threads as a single unit of work, simplifying error handling, improving reliability, and enhancing observability. That is, structured concurrency preserves the readability, maintainability, and observability of single-threaded code.
TrafficResponse retrieveData() throws ExecutionException, InterruptedException {
// CREATE INDEPENDENT TASK VIA THE EXECUTORSERVICE
Future<TrafficData> highwayTask =
this.executorService.submit(this::retrieveHighwayData);
Future<TrafficData> localTask =
this.executorService.submit(this::retrieveLocalData);
// WAIT FOR THE TASKS TO FINISH
TrafficData highwayTraffic = highwayTask.get();
TrafficData localTraffic = localTask.get();
// RETURN COMBINED RESPONSEE
return new TrafficResponse(highwayTraffic, localTraffic);
}TrafficResponse retrieveData() throws ExecutionException, InterruptedException {
// CREATE A NEW SCOPE
try (var scope = new StructuredTaskScope.ShutdownOnFailure) {
// FORK RELATED SUBTASKS
Supplier<TrafficData> highwaySubtask = scope.fork(this::retrieveHighwayData);
Supplier<TrafficData> localSubtask = scope.fork(this::retrieveLocalData);
// JOIN ALL TASKS IN THE SCOPE
scope.join()
.throwIfFailed(); // AND PROPAGATE ERRORS
// GET THE RESULTS
return new TrafficResponse(highwaySubtask.get(),
localTraffic.get());
}
}private static final ScopedValue<String> HOMETWON = ScopedValue.newInstance();
ScopedValue.runWhere(USERNAME, "Heidelberg", () -> {
try (var scope = new StructuredTaskScope<String>()) {
scope.fork(() -> doStuff());
// ...
}
});
String doStuff() {
String name = USERNAME.get();
// => "Heidelberg"
}StructuredTaskScope.ShutdownOnFailure shuts down the scope on the first failed subtask. If you require all results (“invoke all”), this policy ensures that other subtasks get discarded if any of them fails.
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {Future<User> mediumUser = scope.fork(() -> getMediumUser());Future<SubscriptionTier> subscriptionTier = scope.fork(() -> getSubscriptionTier());Future<UserInterests> userInterests = scope.fork(() -> getUserInterests());scope.join();scope.throwIfFailed(IllegalArgumentException::new);return new Response(mediumUser.resultNow(),subscriptionTier.resultNow(),userInterests.resultNow());}
StructuredTaskScope.ShutdownOnSuccess captures the first successfully completed subtask result and shuts down the scope afterward. This will interrupt any unfinished threads and wake up the scope’s owner. Choose this policy if you only need the result of a singular subtask (“invoke any”).
public Weather readWeather() throws ExecutionException, InterruptedException, TimeoutException { try (var scope = new StructuredTaskScope.ShutdownOnSuccess<Weather>()){ // need to call fork and pass runnables or callables scope.fork(Weather::readWeatherFromServerA()); scope.fork(Weather::readWeatherFromServerB()); scope.fork(Weather::readWeatherFromServerC()); // now we block, blocking is cheap in virtual threas // so no issue here scope.join(); Weather weather = scope.result(); return weather; }
public Weather readWeather() throws InterruptedException, ExecutionException { try(var scope = new StructuredTaskScope.ShutdownOnSuccess<Weather>()){ Future<Weather> futureA = scope.fork(Weather::readWeatherFromServerA); Future<Weather> futureB = scope.fork(Weather::readWeatherFromServerB); Future<Weather> futureC = scope.fork(Weather::readWeatherFromServerC); scope.join(); System.out.println("futureA.state() = " + futureA.state()); System.out.println("futureB.state() = " + futureB.state()); System.out.println("futureC.state() = " + futureC.state()); var weather = scope.result(); return weather; } }
WARNING: Using incubator modules: jdk.incubator.concurrent futureA.state() = FAILED futureB.state() = FAILED futureC.state() = SUCCESS
public class OrderedSuccessfulScope<T> extends StructuredTaskScope<T> {
private final Queue<Subtask<? extends T>> subtasks = new LinkedTransferQueue<>();
@Override
protected void handleComplete(Subtask<? extends T> subtask) {
if (subtask.state() != Subtask.State.SUCCESS) {
return;
}
subtasks.add(subtask);
}
// ADDITONAL FUNCTIONALITY PROVIDED BY THE CUSTOM SCOPE
public Stream<Subtask<? extends T>> completedSuccessfully() {
super.ensureOwnerAndJoined();
return subtasks.stream();
}
}The super.ensureOwnerAndJoined() call ensures that all forked tasks are joined, and the current thread is the owner of the scope, or it throws an Exception.
Introducing Scoped Values, which enable sharing immutable data within and between threads. They are preferred over thread-local variables, especially when using a large number of virtual threads.
class Handler {final static ThreadLocal<String> userInfo = new ThreadLocal<>();...}userInfo.set("user-name");....Handler.userInfo.get();
Virtual threads are still Threads, though. That means they can use ThreadLocal<T>, but now, the previously mentioned downsides affect your program at another scale. If you think the overhead or lifetime issues and coordination of a few ThreadLocal<T> are problematic when handling a handful of threads, you better don’t think about these downsides when handling 10.000+ virtual threads!
private static final ThreadLocal<Principal> PRINCIPAL_TL = new ThreadLocal(); // - VS - private static final ScopedValue<Principal> PRINCIPAL_SV = ScopedValue.newInstance();
<T> ScopedValue.Carrier where(ScopedValue<T> key, T value)
void run(Runnable op) <R> R call(Callable<? extends R> op)
private static final ScopedValue<String> USER = ScopedValue.newInstance();
ScopedValue.Carrier carrier = ScopedValue.where(USER, "ben");
var boundUser = carrier.get(USER);
// => "ben"
carrier.run(() -> {
var currentUser = USER.get();
// => "ben"
});private static final ScopedValue<String> USER = ScopedValue.newInstance();
// CREATE TWO CARRIERS WITH DIFFERENT VALUES IN THE SAME THREAD
ScopedValue.Carrier carrierForBen = ScopedValue.where(USER, "ben");
ScopedValue.Carrier carrierForBob = ScopedValue.where(USER, "bob");
// USE THE SAME RUNNABLE FOR EACH CARRIER
Runnable r = () -> {
var currentUser = USER.get();
System.out.println("Current: " + currentUser);
};
// RUN RUNNABLE WITH DIFFERENT CARRIERS
carrierForBen.run(r);
// => "Current: ben"
carrierForBob.run(r);
// => "Current: bob"
carrierForBen.run(r);
// => "Current: ben"