结合使用Springboot JMS 与 Amazon SQS 标准队列

àì夳堔傛蜴生んèń 2023-06-16 10:50 7阅读 0赞

1. 首先，来了解下[JMS的一些概念][JMS]，[AWS官网例子][AWS]。关于SQS的使用总结可以参考：[https://blog.csdn.net/BAStriver/article/details/103262276][https_blog.csdn.net_BAStriver_article_details_103262276]

2. 假设我们已经创建好一个springboot项目。那么，以下是个很简单的入门使用总结。

1) 先引入最新的依赖包：

<dependency>
    	<groupId>org.springframework</groupId>
    	<artifactId>spring-jms</artifactId>
    	<version>4.3.6.RELEASE</version>
    </dependency>
    <dependency>
    	<groupId>com.amazonaws</groupId>
    	<artifactId>amazon-sqs-java-messaging-lib</artifactId>
    	<version>1.0.8</version>
    </dependency>
    <dependency>
    	<artifactId>iam</artifactId>
    	<groupId>software.amazon.awssdk</groupId>
    	<version>2.3.7</version>
    </dependency>

2) JMS配置类，初始化一些JMS配置：

package com.bas.configuration;
    
    import com.amazon.sqs.javamessaging.ProviderConfiguration;
    import com.amazon.sqs.javamessaging.SQSConnectionFactory;
    import com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
    import com.amazonaws.services.sqs.AmazonSQSClientBuilder;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.jms.annotation.EnableJms;
    import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
    import org.springframework.jms.support.destination.DynamicDestinationResolver;
    
    import javax.jms.Session;
    
    @Configuration
    @EnableJms
    @Profile({"stg", "dev", "prod"})
    public class JmsConfiguration {
    
        private final AmazonSQSClientBuilder sqsClientBuilder;
        private final SQSConnectionFactory connectionFactory;
    
        public JmsConfiguration(@Value("${s3.regionKey}") String s3Region ) {
            sqsClientBuilder = AmazonSQSClientBuilder.standard()
                    .withCredentials(DefaultAWSCredentialsProviderChain.getInstance())
                    .withRegion(s3Region);
            connectionFactory = new SQSConnectionFactory(new ProviderConfiguration(),this.sqsClientBuilder);
        }
    
        @Bean
        public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(){
            DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
    
            factory.setConnectionFactory(this.connectionFactory);
            factory.setDestinationResolver(new DynamicDestinationResolver());
    
            //The values we provided to Concurrency show that we will create a minimum of 3 listeners that will scale up to 10 listeners
            factory.setConcurrency("3-10");
    
            /**
             *  SESSION_TRANSACTED
             *  CLIENT_ACKNOWLEDGE : After the client confirms, the client must call the acknowledge method of javax.jms.message after receiving the message. After the confirmation, the JMS server will delete the message
             *  AUTO_ACKNOWLEDGE : Automatic acknowledgment, no extra work required for client to send and receive messages
             *  DUPS_OK_ACKNOWLEDGE : Allow the confirmation mode of the replica. Once a method call from the receiving application returns from the processing message, the session object acknowledges the receipt of the message; and duplicate acknowledgments are allowed. This pattern is very effective when resource usage needs to be considered.
             */
            factory.setSessionAcknowledgeMode(Session.CLIENT_ACKNOWLEDGE);
    
            return factory;
        }
    }

3) 以消费者为例：

package com.bas.configuration;
    
    import com.amazon.sqs.javamessaging.message.SQSTextMessage;
    import com.bas.service.SqsService;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jms.annotation.JmsListener;
    import org.springframework.stereotype.Component;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    
    @Component
    public class SqsConsumer {
    
        @Autowired
        private SqsService sqsService;
    
        @JmsListener(destination = "${sqs.queueName}")
        public void consumePortfolioMessage(Message message) {
            try {
                SQSTextMessage textMessage = (SQSTextMessage) message;
                sqsService.handleMessage(textMessage.getText());
                message.acknowledge(); // 删除消息
            } catch (JMSException e) {
    
            }
    
        }
    }

因为设置的是客户端确认模式，所以要记得调用acknowledge()删除sqs消息。

![20191126113032582.png][]

如果在消费消息的时候需要将消息转成实体对象，可以参考：[springboot + aws +sqs + jms(简单托管消息队列)][springboot _ aws _sqs _ jms]。

@JmsListener(destination = "${sqs.queueName}")
    public void test(@Payload final Message<MessageBody> message) {
    	System.out.println("Listening {} in queue test"+ message.getPayload().getMessageId());
        // getMessageId() 是MessageBody的getter
    }

**注**：

1. 如果遇到：...jms.support.converter.MessageConversionException: Could not find type id property。

出现这个问题是因为JMS配置里面的转换器不能正常处理消息转成我们设置的MessageBody实体类。

2. 如果要中途关闭listener可以参考：[https://segmentfault.com/n/1330000015554448][https_segmentfault.com_n_1330000015554448]

3. 如果遇到：...s3.model.S3Exception: Please reduce your request rate. (Service: S3, Status Code: 503...

出现这个问题是因为请求速率太频繁了超过了限制，可以参考：[Amazon S3；状态代码：503；错误代码：SlowDown][Amazon S3_503_SlowDown]

[JMS]: https://www.cnblogs.com/progor/p/10877658.html
[AWS]: https://docs.aws.amazon.com/zh_cn/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-jms-code-examples.html
[https_blog.csdn.net_BAStriver_article_details_103262276]: https://blog.csdn.net/BAStriver/article/details/103262276
[20191126113032582.png]: https://img-blog.csdnimg.cn/20191126113032582.png
[springboot _ aws _sqs _ jms]: https://www.jianshu.com/p/e9e51eaa1288
[https_segmentfault.com_n_1330000015554448]: https://segmentfault.com/n/1330000015554448
[Amazon S3_503_SlowDown]: https://aws.amazon.com/cn/premiumsupport/knowledge-center/s3-resolve-503-slowdown-throttling/