De kracht van Request Scoped Beans benutten: @RequestScope in een niet-webgebaseerd verzoek

Spring, het veelzijdige framework, biedt ontwikkelaars een reeks opmerkelijke functies voor het beheren van de levenscyclus van bonen. Typisch behandelt Spring de levenscyclus van bonen wanneer we annotaties gebruiken zoals @Service of @Component. Als de Spring Container opstart, worden deze beans aangemaakt en bij het afsluiten worden ze netjes vernietigd. Maar buiten deze bekende methodes, bestaat er een schat aan mogelijkheden om de levenscyclus van bonen te verfijnen.

In deze blogpost duiken we in een aspect van Spring dat bekend staat als "Spring Bean Scopes". Specifiek verkennen we de veelzijdige mogelijkheden van een request scoped bean en ontdekken we hoe deze kan worden gebruikt buiten de grenzen van een webgebaseerd verzoek.

De uitdaging van RequestScope aangaan

RequestScope werkt naadloos binnen een webgebaseerde context wanneer het wordt afgehandeld door de Spring DispatcherServlet. Er is echter een probleem wanneer je een request scoped bean probeert te benaderen buiten de grenzen van een web request. Je kunt een uitzondering tegenkomen die lijkt op de volgende:
‍
java.lang.IllegalStateException: No thread-bound request found: Bedoel je request attributen buiten een echt webverzoek, of het verwerken van een verzoek buiten de oorspronkelijk ontvangende thread? Als je daadwerkelijk werkt binnen een webverzoek en toch deze melding krijgt, draait je code waarschijnlijk buiten DispatcherServlet/DispatcherPortlet: Gebruik in dat geval RequestContextListener of RequestContextFilter om het huidige verzoek bloot te leggen.

Toch is deze beperking geen fout in het ontwerp van Spring. Het framework legt de RequestScope niet bloot omdat het niet kan bepalen wanneer een verzoek begint en eindigt buiten de DispatcherServlet. In plaats daarvan verwacht het van ontwikkelaars dat ze deze verantwoordelijkheid op zich nemen, en we kunnen dit bereiken door de volgende stappen uit te voeren:

Maak een CustomRequestScopeAttr (Grotendeels gekopieerd van blogpost van Pranav Maniar)

import org.springframework.web.context.request.RequestAttributes;

import java.util.HashMap;
import java.util.Map;

public class CustomRequestScopeAttr implements RequestAttributes {
    private final Map requestAttributeMap = new HashMap<>();

    @Override
    public Object getAttribute(String name, int scope) {
        if (scope == RequestAttributes.SCOPE_REQUEST) {
            return requestAttributeMap.get(name);
        }

        return null;
    }

    @Override
    public void setAttribute(String name, Object value, int scope) {
        if (scope == RequestAttributes.SCOPE_REQUEST) {
            requestAttributeMap.put(name, value);
        }
    }

    @Override
    public void removeAttribute(String name, int scope) {
        if (scope == RequestAttributes.SCOPE_REQUEST) {
            requestAttributeMap.remove(name);
        }
    }

    @Override
    public String[] getAttributeNames(int scope) {
        if (scope == RequestAttributes.SCOPE_REQUEST) {
            return requestAttributeMap
                    .keySet()
                    .toArray(new String[0]);
        }

        return new String[0];
    }
 // todo implement other methods (not used. just return null)
} 

Dit zal alleen beschikbaar zijn en werken voor request scoped beans Beans en niet ook voor andere scopes.
‍
Vervolgens moeten we handmatig de RequestAttributes instellen. Dit moet gedaan worden waar logischerwijs ons request scoped proces start. Dit wordt gedaan door aan te roepen:

RequestContextHolder.setRequestAttributes(nieuw CustomRequestScopeAttr())  

Zodra het request scoped proces is voltooid, resetten we de RequestAttributes door aan te roepen:

RequestContextHolder.resetRequestAttributes();

Door deze methoden aan te roepen RequestContextHolder.setRequestAttributes(new CustomRequestScopeAttr()) om het proces te starten en RequestContextHolder.resetRequestAttributes() om het te beëindigen - nemen we de controle over het bepalen wanneer een verzoek begint en eindigt.

In de volgende secties zullen we verschillende scenario's verkennen om te illustreren hoe we het begin en einde van een verzoek effectief kunnen bepalen.

Request Scoped Beans gebruiken in @Async

Het tegenkomen van een @Async annotatie in een webgebaseerd verzoek kan leiden tot een IllegalStateException bij een poging om toegang te krijgen tot een request scoped bean. Om dit op te lossen, moeten we de RequestAttributes instellen bij het aanmaken van de async thread. Hiervoor maken we een aangepaste AsyncConfiguration om het aanmaken van async threads te beheren, waarbij we de RequestAttributes dienovereenkomstig instellen.

Hieronder volgt een voorbeeld van een AsyncConfiguration:

@EnableAsync
@EnableScheduling
public class AsyncConfiguration implements AsyncConfigurer {
    private final TaskExecutionProperties taskExecutionProperties;

    @Override
    @Bean(name = "taskExecutor")
    public Executor getAsyncExecutor() {
        log.debug("Creating Async Task Executor");

        return getExecutor(
                taskExecutionProperties.getPool(),
                taskExecutionProperties.getThreadNamePrefix()
        );
    }
    private Executor getExecutor(TaskExecutionProperties.Pool pool, String threadNamePrefix) {
        ContextAwarePoolExecutor executor = new ContextAwarePoolExecutor();

        executor.setCorePoolSize(pool.getCoreSize());
        executor.setMaxPoolSize(pool.getMaxSize());
        executor.setQueueCapacity(pool.getQueueCapacity());
        executor.setThreadNamePrefix(threadNamePrefix);

        return executor;
    }

Wanneer je een @Async blok invoert, wordt getAsyncExecutor() aangeroepen en "taskExecutor" is de standaardnaam. Je kunt ook een executornaam doorgeven in de @Async annotatie. De configuratie wordt dan doorgegeven aan onze aangepaste ContextAwarePoolExecutor, die is gedefinieerd als:

import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.context.request.RequestContextHolder;

import java.util.concurrent.Callable;
import java.util.concurrent.Future;

public class ContextAwarePoolExecutor extends ThreadPoolTaskExecutor {
    /**
     * @param task the {@code Callable} to execute (never {@code null}) - is the actual method we want to call
     */
    @Override
    public  Toekomst submit(Oproepbaar task) {
        return super.submit(
                new ContextAwareCallable(RequestContextHolder.currentRequestAttributes(), task)
        );
    }

    @Override
    public  LuisterbareToekomst submitListenable(Callable task) {
        return super.submitListenable(
                new ContextAwareCallable(RequestContextHolder.currentRequestAttributes(), task)
        );
    }
}

De ContextAwarePoolExecutor handelt de eigenlijke uitvoering van de nieuw aangemaakte thread af en zorgt ervoor dat de RequestAttributen juist zijn ingesteld:

importeer org.springframework.web.context.request.RequestAttributes;
importeer org.springframework.web.context.request.RequestContextHolder;

import java.util.concurrent.Callable;

openbare klasse ContextAwareCallable implementeert Callable {
    private final CustomRequestScopeAttr requestAttributes;
    private Callable taak;

    openbare ContextAwareCallable(RequestAttributes requestAttributes, Callable task) {
        this.task = task;
        this.requestAttributes = cloneRequestAttributes(requestAttributes);
    }

    @Override
    public T call() throws Exception {
        try {
            RequestContextHolder.setRequestAttributes(requestAttributes);
            return task.call();
        } finally {
            RequestContextHolder.resetRequestAttributes();
        }
    }

    /**
     * this is needed, because once the main thread is finished, the object may get garbage collected, even if the async thread is not finished
     *
     * @param requestAttributes
     * @return
     */
    private CustomRequestScopeAttr cloneRequestAttributes(RequestAttributes requestAttributes) {
        CustomRequestScopeAttr clonedRequestAttribute = null;

        try {
            clonedRequestAttribute = new CustomRequestScopeAttr();

            for (String name : requestAttributes.getAttributeNames(RequestAttributes.SCOPE_REQUEST)) {
                clonedRequestAttribute.setAttribute(
                        name,
                        requestAttributes.getAttribute(name, RequestAttributes.SCOPE_REQUEST),
                        RequestAttributes.SCOPE_REQUEST
                );
            }
            return clonedRequestAttribute;
        } catch (Exception e) {
            return new CustomRequestScopeAttr();
        }
    }
}

De ContextAwareCallable klasse implementeert de Callable interface en handelt de uitvoering van de nieuwe thread af. In de call() methode stellen we de RequestAttributen in om de start van het verzoek te "markeren". Aangezien we de RequestAttributes doorgeven van de hoofd thread, moeten we ze klonen voor het instellen. Dit voorkomt problemen waarbij de hoofd thread mogelijk eerder eindigt dan de async thread, wat leidt tot attribuut-afvalverzameling. In het slot-blok van de call() methode "markeren" we het einde van het verzoek door de RequestAttributen te resetten. Deze aanpak maakt cascadering van meerdere async-aanroepen mogelijk, omdat verzoekattributen naar beneden worden doorgegeven en altijd worden gekloond voor elke nieuwe thread.

Door dit mechanisme te gebruiken, kun je naadloos @Async annotaties gebruiken met behoud van de functionaliteit van request scoped beans.

Gebruik van Request Scoped Beans met Pub/Sub

De firewallregel voor de verbinding met het backend is alleen van toepassing op een specifieke tag, die eruitziet als een knooppunt. Bijvoorbeeld: "gke-staging-456b4340-node"Dit is echter een Network Tag, die op elke Compute Instance van het cluster staat. Bij het werken met Pub/Sub consume events, die verzoeken zijn van een wachtrij in plaats van webgebaseerde interacties, vereist het gebruik van request scoped beans het instellen van RequestAttributes. Gelukkig is het definiëren van het begin en einde van een verzoek in een Pub/Sub "context" eenvoudig, omdat we duidelijk kunnen zien wanneer een gesprek begint en eindigt. In ons geval gebruiken we Spring Cloud GCP Pub/Sub.

We hebben een PubSubConsumer klasse die er op de een of andere manier zo uitziet:

@Slf4j
openbare abstracte klasse PubSubConsumer { 
...
    public MessageReceiver receiver() {
        return (PubsubMessage message, AckReplyConsumer consumer) -> {
            try {
                String messageString = parseMessageToString(message, consumer);
                if (messageString == null) {
                    return;
                }

                startConsumeProcess(messageString);

                consumer.ack();
            } catch (NackException e) {
                // we are fine. just nack and try again
                log.info("received nack exception. we will nack this queue entry", e);
                consumer.nack();
            } catch (AckException e) {
                // we are fine. we can ack this one
                log.info("received ack exception. we will ack this queue entry", e);
                consumer.ack();
            } catch (Exception e) {
                // we are not fine
                log.error("error while receiving message from subscription {}", subscription, e);
                consumer.nack();
            } finally {
                RequestContextHolder.resetRequestAttributes();
            }
        };
    }

    protected T parseStringToPayloadType(String messageString) {
        try {
            return objectMapper.readValue(messageString, payloadType);
        } catch (IOException e) {
           ...
        }
    }

    protected String parseMessageToString(PubsubMessage message, AckReplyConsumer consumer) {
        log.info("receive message from subscription {} with payload {}", subscription, message);
        if (message == null || message.getData() == null) {
            ...
        }
        return message.getData().toStringUtf8();
    }

    /**
     * actual consumer process logic.
     *
     * @param messageString String content of a message.
     * @throws Exception
     */
    protected void startConsumeProcess(String messageString) throws Exception {
        RequestContextHolder.setRequestAttributes(new CustomRequestScopeAttr());

        T payload = parseStringToPayloadType(messageString);

        setContextVariables(payload);

        consume(payload);
    }
...
}

Bij het starten van het consume-proces in de methode startConsumeProcess() stellen we de RequestScope-attributen in. In het slot-blok van de eigenlijke Receiver resetten we de RequestAttributes. Het hele Pub/Sub-evenement, van het begin van de consumer tot het einde, vormt een compleet verzoek, waardoor het beheer van request scoped beans naadloos werkt binnen deze context.

Zelfs als je een asynchrone methode aanroept binnen de consumer, blijft deze aanpak effectief. Zorg ervoor dat je de wijzigingen uit het hoofdstuk "Async" implementeert om asynchrone scenario's correct af te handelen. Door deze strategieën te combineren, kun je vol vertrouwen gebruik maken van request scoped beans in Pub/Sub events, wat robuuste en efficiënte verwerking van berichten binnen je Spring applicatie mogelijk maakt.

Omgaan met ParallelStream in Java

Tot nu toe heb ik helaas nog geen haalbare / generieke oplossing gevonden om toegang te krijgen tot een request scoped bean in een java ParallelStream. Het onderliggende probleem ligt in het gebruik van een gemeenschappelijke Fork/Join Pool door Java streams voor parallellisatie. Deze threads worden niet aangemaakt of geconfigureerd door Spring of de ontwikkelaar, in tegenstelling tot wat we deden in de "Async" sectie. Je zou handmatig de request attributen voor elke ParallelStream kunnen instellen, maar dit is geen praktische oplossing, vooral als je al te maken hebt met meerdere ParallelStreams, wat kan leiden tot mogelijke vergissingen.

Een andere benadering zou zijn om een aangepaste ForkJoinPool te maken en deze te gebruiken voor parallelstreamen. Dit kan er ongeveer zo uitzien (gekopieerd van hier):
‍

final int parallelism = 4;
ForkJoinPool forkJoinPool = null;
try {
    forkJoinPool = new ForkJoinPool(parallelism);
    final List primes = forkJoinPool.submit(() ->
        // Parallel task here, for example
        IntStream.range(1, 1_000_000).parallel()
                .filter(PrimesPrint::isPrime)
                .boxed().collect(Collectors.toList())
    ).get();
    System.out.println(primes);
} catch (InterruptedException | ExecutionException e) {
    throw new RuntimeException(e);
} finally {
    if (forkJoinPool != null) {
        forkJoinPool.shutdown();
    }
}

Deze aanpak vereist het expliciet aanmelden van taken bij de aangepaste forkJoinPool elke keer dat je parallelle bewerkingen wilt uitvoeren, waardoor het gebruik van ParallelStreams onpraktisch wordt.

We hebben besloten om de beperking te accepteren dat we geen request-scoped beans kunnen gebruiken in ParallelStreams. In plaats daarvan gebruiken we alternatieve strategieën om onze doelen te bereiken zonder te vertrouwen op request-scoped beans in deze parallelle uitvoeringsscenario's. Hoewel deze beperking bepaalde beperkingen met zich mee kan brengen, kunnen we ParallelStreams effectief gebruiken binnen onze Spring applicatie door deze te begrijpen en er omheen te werken.

Samenvatting

Request scoped beans in Spring bieden een aanzienlijke flexibiliteit en breiden hun gebruik uit tot buiten de traditionele webgebaseerde requests. Wanneer deze beans echter buiten de standaard webcontext worden gebruikt, moeten ontwikkelaars de verantwoordelijkheid nemen voor het definiëren van het begin en einde van elk proces. Spring kan dit niet automatisch regelen in aangepaste omgevingen of procesflows. In verschillende scenario's, zoals asynchrone verwerking, pub/sub gebeurtenissen en aangepaste threadpools, hebben we effectieve technieken gedemonstreerd voor het omgaan met request scoped beans. De enige beperking die we tot nu toe zijn tegengekomen is met ParallelStreams, waar een generieke oplossing voor toegang tot request-scoped beans nog steeds ontbreekt. Ondanks deze beperking kunnen ontwikkelaars, door de beperkingen te begrijpen en alternatieve strategieën toe te passen, het volledige potentieel van request-scoped beans in hun Spring-applicaties benutten.

Kai Müller
Software-ingenieur