I have a requirement in which I want to consume multiple SQS topic from a single java process. Below is the sample code:
public class MultiTopicSQSConsumer {
Connection connection;
private void init(){
try {
AwsCredentialsProvider credentialsProvider = StaticCredentialsProvider.create(AwsBasicCredentials.create("accessKey", "secretKey"));
SqsClient sqsClient = SqsClient.builder()
.region(Region.of("regionName"))
.credentialsProvider(credentialsProvider).build();
SQSConnectionFactory connectionFactory = new SQSConnectionFactory(
new ProviderConfiguration(),
sqsClient
);
connection = connectionFactory.createConnection();
connection.start();
} catch (Exception ex){
throw new RuntimeException("Failed to create connection. Error - " + ex.getMessage() );
}
}
private void createConsumer(String queueName){
try {
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Queue queue = session.createQueue(queueName);
MessageConsumer consumer = session.createConsumer(queue);
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(javax.jms.Message message) {
// handle message
}
});
} catch (Exception ex){
throw new RuntimeException("Failed to create session. Error - " + ex.getMessage());
}
}
}
The code works fine. But in case of high load like 10 consumers(each for different queue) for the message listener hangs and recovers after long time (hours after receiving the message).
Is it a expected behavior or there is something wrong with the implementation.