AWS SQS – MessageConsumer smette di ricevere messaggi dopo un po'
La mia applicazione registra un listener in una coda SQS (la coda stessa è popolata da un argomento SNS). Quando avvio l'applicazione, il consumer del messaggio funziona come previsto ma dopo un po' smette di ricevere i messaggi. Può essere che il consumatore si stia spegnendo dopo un po'?
Suggerimenti o commenti sarebbero molto apprezzati.
Collegamento SQS:
@Bean public SQSConnection amazonSQSConnection( @Value("${aws.access.key}") String accessKey, @Value("${aws.secret.key}") String secretKey) throws JMSException { BasicAWSCredentials basicAWSCredentials = new BasicAWSCredentials(accessKey, secretKey); AmazonSQSClientBuilder client = AmazonSQSClientBuilder .standard() .withRegion(Regions.GovCloud) .withCredentials(new AWSStaticCredentialsProvider(basicAWSCredentials)); SQSConnectionFactory connectionFactory = new SQSConnectionFactory(new ProviderConfiguration(), client); return connectionFactory.createConnection(); }
Consumatore:
@Bean public MessageConsumer workOrderChangeConsumer( SQSConnection connection, WorkOrderKittingService workOrderKittingService, AuthenticationProvider authProvider, @Value("${app.user.name}") String appUserName, @Value("${aws.sqs.workorder.change.queue}") String woChangeQueue) throws JMSException { Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE); Queue queue = session.createQueue(woChangeQueue); WorkOrderChangeIngestor workOrderChangeIngestor = new WorkOrderChangeIngestor( workOrderKittingService, authProvider, appUserName); MessageConsumer consumer = session.createConsumer(queue); consumer.setMessageListener(workOrderChangeIngestor); connection.start(); return consumer; }
Risposta
Stai cercando di gestire da solo il ciclo di vita della connessione.
Ti consiglio di lasciare che sia la primavera a gestirlo per te, usando spring-cloud-aws
https://docs.spring.io/spring-cloud-aws/docs/2.2.3.RELEASE/reference/html/#receiving-a-message
Puoi creare un listener tramite le annotazioni:
@Component public class MyMessageHandler { @SqsListener("queueName") void handle(String message) { ... throw new MyException("something went wrong"); } @MessageExceptionHandler(MyException.class) void handleException(MyException e) { ... } }