首页 > Java > java教程 > 正文

使用 Amazon SQS 构建 Spring Boot 消费者应用程序:使用云开发套件 (CDK) 设置基础设施

Susan Sarandon
发布: 2024-10-07 22:08:30
原创
504 人浏览过
第 017 天 - 100 天AWSIaCDevops 挑战

今天,在我的 100 天代码挑战系列中,我将向您展示如何使用 Amazon SQS 解耦使用 springboot 开发的微服务。

什么是亚马逊 SQS?

Amazon SQS(简单队列服务) 是一项云服务,可帮助应用程序通过 sendyg 进行通信,在队列中存储和接收消息。它就像一条等待队列,等待消费者准备好处理它们为止。这可以防止系统失控并确保不会丢失消息。

使用 Springboot 应用程序使用 SQS 消息

演示如何通过创建一个处理来自 SQS 队列的每条消息的 Spring Boot 应用程序来使用 SQS 消息。使用 CDK (Java) 构建的基础设施将包括:

  • VPC,具有公共 子网,用于托管将在其中运行 Spring Boot 应用程序的 EC2 实例。
  • 用于 EC2 实例访问互联网并下载依赖项的互联网网关
  • 用于消息存储的 SQS 队列死信队列
  • 用于托管 SpringBoot 应用程序的 EC2 实例
  • IAM 角色,允许 EC2 实例从 SQS 队列检索消息(非常重要)。

创建基础设施

使用 CDK (Java) 设置必要的基础设施

Building a Spring Boot Consumer Application with Amazon SQS: Setup Infrastructure Using Cloud Development Kit (CDK)

VPC 和子网互联网网关


public class NetworkContruct extends Construct {
  private final IVpc vpc;
  public NetworkContruct(Construct scope, String id, StackProps props) {
    super(scope, id);
    this.vpc =
        new Vpc(
            this,
            "VpcResource",
            VpcProps.builder()
                .vpcName("my-vpc")
                .enableDnsHostnames(true)
                .enableDnsSupport(true)
                .createInternetGateway(true)
                .ipProtocol(IpProtocol.IPV4_ONLY)
                .ipAddresses(IpAddresses.cidr("10.0.0.1/16"))
                .maxAzs(1)
                .subnetConfiguration(
                    List.of(
                        SubnetConfiguration.builder()
                            .name("Public-Subnet")
                            .mapPublicIpOnLaunch(true)
                            .subnetType(SubnetType.PUBLIC)
                            .build()))
                .build());
  }
  public IVpc getVpc() {
    return vpc;
  }
}


登录后复制

此 cdk 构造将创建:
??名为 my-vpc 的 VPC 并启用 DNS 主机名。
??名为 Public-Subnet 的公共子网,允许资源附加公共 IP(如果配置了公共 IP)。
??用于启用互联网流量的互联网网关。

SQS队列和死信队列


public class QueueConstruct extends Construct {
  private final IQueue queue;
  public QueueConstruct(Construct scope, String id, IVpc vpc, StackProps props) {
    super(scope, id);
    IQueue dlq =
        new Queue(
            this,
            "DeadLetterQueue",
            QueueProps.builder()
                .deliveryDelay(Duration.millis(0))
                .retentionPeriod(Duration.days(10))
                .queueName("my-queue-dlq")
                .build());
    DeadLetterQueue deadLetterQueue = DeadLetterQueue.builder()
        .queue(dlq)
        .maxReceiveCount(32)
        .build();

    this.queue =
        new Queue(
            this,
            "SQSQueueResource",
            QueueProps.builder()
                .queueName("my-queue")
                .retentionPeriod(Duration.minutes(15))
                .visibilityTimeout(Duration.seconds(90))
                .deadLetterQueue(deadLetterQueue)
                .build());
  }

  public IQueue getQueue() {
    return queue;
  }
}


登录后复制

上述 CDK 构造将创建以下资源:

  • 一个名为 my-queue 的队列,它将在 Spring Boot 应用程序中使用。
  • 一个名为 my-queue-dlq 的死信队列,它捕获失败的消息以便稍后可以分析和修复它们,而不会阻塞主队列。

用于托管 Spring Boot 应用程序的 EC2 实例


// ComputerProps.java
public record ComputerProps(IVpc vpc, String sqsQueueArn) {}

// ComputerConstruct.java
public class ComputerConstruct extends Construct {
  private final IInstance computer;
  public ComputerConstruct(
      Construct scope, String id, ComputerProps computerProps, StackProps props) {
    super(scope, id);
    SecurityGroup securityGroup =
        new SecurityGroup(
            this,
            "WebserverSGResource",
            SecurityGroupProps.builder()
                .allowAllOutbound(true)
                .securityGroupName("Webserver-security-group")
                .disableInlineRules(true)
                .vpc(computerProps.vpc())
                .description("Allow trafic from/to webserver instance")
                .build());

    securityGroup.addIngressRule(Peer.anyIpv4(), Port.SSH, "Allow ssh traffic");
    securityGroup.addIngressRule(Peer.anyIpv4(), Port.tcp(8089), "Allow traffic from 8089 port");

    KeyPair keyPair =
        new KeyPair(
            this,
            "KeyPairResource",
            KeyPairProps.builder()
                .keyPairName("ws-keypair")
                .account(Objects.requireNonNull(props.getEnv()).getAccount())
                .type(KeyPairType.RSA)
                .format(KeyPairFormat.PEM)
                .build());

    new CfnOutput(
        this, "KeyPairId", CfnOutputProps.builder().value(keyPair.getKeyPairId()).build());

    Instance ec2Instance =
        new Instance(
            this,
            "WebServerInstanceResource",
            InstanceProps.builder()
                .securityGroup(securityGroup)
                .keyPair(keyPair)
                .instanceName("Webserver-Instance")
                .machineImage(
                    MachineImage.lookup(
                        LookupMachineImageProps.builder()
                            .name("*ubuntu*")
                            .filters(
                                Map.ofEntries(
                                    Map.entry("image-id", List.of("ami-0e86e20dae9224db8")),
                                    Map.entry("architecture", List.of("x86_64"))))
                            .windows(false)
                            .build()))
                .vpc(computerProps.vpc())
                .role(buildInstanceRole(computerProps))
                .instanceType(InstanceType.of(InstanceClass.T2, InstanceSize.MICRO))
                .associatePublicIpAddress(true)
                .blockDevices(
                    List.of(
                        BlockDevice.builder()
                            .mappingEnabled(true)
                            .deviceName("/dev/sda1")
                            .volume(
                                BlockDeviceVolume.ebs(
                                    10,
                                    EbsDeviceOptions.builder()
                                        .deleteOnTermination(true)
                                        .volumeType(EbsDeviceVolumeType.GP3)
                                        .build()))
                            .build()))
                .userDataCausesReplacement(true)
                .vpcSubnets(SubnetSelection.builder().subnetType(SubnetType.PUBLIC).build())
                .build());

    UserData userData = UserData.forLinux();
    userData.addCommands(readFile("./webserver-startup.sh"));

    ec2Instance.addUserData(userData.render());

    this.computer = ec2Instance;
  }

  public IInstance getComputer() {
    return computer;
  }

  private String readFile(String filename) {

    InputStream scriptFileStream = getClass().getClassLoader().getResourceAsStream(filename);

    try {
      assert scriptFileStream != null;
      try (InputStreamReader isr = new InputStreamReader(scriptFileStream, StandardCharsets.UTF_8);
          BufferedReader br = new BufferedReader(isr)) {
        StringBuilder content = new StringBuilder();
        String line;
        while ((line = br.readLine()) != null) {
          content.append(line).append("\n");
        }
        return content.toString();
      }
    } catch (IOException e) {
      throw new RuntimeException(e.getMessage());
    }
  }

  private IRole buildInstanceRole(ComputerProps props) {
    return new Role(
        this,
        "WebserverInstanceRoleResource",
        RoleProps.builder()
            .roleName("webserver-role")
            .assumedBy(new ServicePrincipal("ec2.amazonaws.com"))
            .path("/")
            .inlinePolicies(
                Map.ofEntries(
                    Map.entry(
                        "sqs",
                        new PolicyDocument(
                            PolicyDocumentProps.builder()
                                .assignSids(true)
                                .statements(
                                    List.of(
                                        new PolicyStatement(
                                            PolicyStatementProps.builder()
                                                .effect(Effect.ALLOW)
                                                .actions(
                                                    List.of(
                                                        "sqs:DeleteMessage",
                                                        "sqs:ReceiveMessage",
                                                        "sqs:SendMessage",
                                                        "sqs:GetQueueAttributes",
                                                        "sqs:GetQueueUrl"))
                                                .resources(List.of(props.sqsQueueArn()))
                                                .build())))
                                .build()))))
            .build());
  }
}


登录后复制

上述 CDK 构造将创建以下资源:

  • 名为 Webserver-security-group 的安全组,允许端口 22 上的 SSH 连接的入站流量,并允许 端口 8089(应用程序连接端口)上的入站流量。
  • 名为 ws-keypair 的密钥对,将用于通过 SSH 连接到应用程序主机。由于我们使用 CDK 来构建基础设施,如果部署后需要下载私钥(PEM 文件),请参阅我之前的文章如何在 Cloudformation 或 CDK 堆栈创建后检索私钥文件 PEM[↗]。
  • 名为 Webserver-Instance 的 Ec2 实例。
  • 名为 webserver-role 的 Ec2 实例的 IAM 角色,它允许 Ec2 实例上托管的 spring Boot 应用程序与 Amazon SQS 队列(已创建)建立连接并执行操作:删除消息、接收消息、发送消息,获取队列属性并获取队列Url。

创建堆栈


// MyStack.java
public class MyStack extends Stack {
  public MyStack(final Construct scope, final String id, final StackProps props) {
    super(scope, id, props);
    IVpc vpc = new NetworkContruct(this, "NetworkResource", props).getVpc();
    IQueue queue = new QueueConstruct(this, "QueueResource", vpc, props).getQueue();
    IInstance webserver =
        new ComputerConstruct(
                this, "ComputerResource", new ComputerProps(vpc, queue.getQueueArn()), props)
            .getComputer();
  }
}

// Day17App.java
public class Day017App {
  public static void main(final String[] args) {
    App app = new App();
    new MyStack(app,"Day017Stack",
        StackProps.builder()
                .env(
                    Environment.builder()
                        .account(System.getenv("CDK_DEFAULT_ACCOUNT"))
                        .region(System.getenv("CDK_DEFAULT_REGION"))
                        .build())
                .build());
    app.synth();
  }
}


登录后复制

创建SpringBoot消费者应用程序

为了让事情变得简单并避免让我的生活变得复杂,我将使用 Spring Cloud AWS Docs[↗]

Spring Cloud AWS 简化了在 Spring 框架和 Spring Boot 应用程序中使用 AWS 托管服务。它提供了一种使用众所周知的 Spring 习惯用法和 API 与 AWS 提供的服务进行交互的便捷方式。

为了在我的项目中配置SQS服务,我将在配置类中添加以下bean:


@Configuration
public class ApplicationConfiguration {
    @Bean
    public AwsRegionProvider customRegionProvider() {
        return new InstanceProfileRegionProvider();
    }
    @Bean
    public AwsCredentialsProvider customInstanceCredProvider() {
        return  InstanceProfileCredentialsProvider.builder()
                .build();
    }
}


登录后复制

以及捕获所有新消息并打印其内容的侦听器。


@Slf4j
@Component
public class ExampleSQSConsumer {
    @SqsListener(queueNames = { "my-queue" }) // ??
    public void listen(String payload) {
        log.info("*******************  SQS Payload ***************");
        log.info("Message Content: {}", payload);
        log.info("Received At: {}", Date.from(Instant.now()));
        log.info("************************************************");
    }
}


登录后复制

您可以在我的 GitHub 存储库中找到完整的项目[↗]

部署

⚠️⚠️ 在运行部署命令之前,请确保您的主机上安装了 java。我在 MacO 下使用 Java 21 来构建这个基础设施。

在任意位置打开终端并运行以下命令:


<p>git clone https://github.com/nivekalara237/100DaysTerraformAWSDevops.git<br>
cd 100DaysTerraformAWSDevops/day_017<br>
cdk bootstrap --profile cdk-user<br>
cdk deploy --profile cdk-user Day017Stack</p>

登录后复制




结果

Building a Spring Boot Consumer Application with Amazon SQS: Setup Infrastructure Using Cloud Development Kit (CDK)


您可以在 GitHub Repo 上找到完整的源代码↗

以上是使用 Amazon SQS 构建 Spring Boot 消费者应用程序:使用云开发套件 (CDK) 设置基础设施的详细内容。更多信息请关注PHP中文网其他相关文章!

来源:dev.to
本站声明
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系admin@php.cn
作者最新文章
热门教程
更多>
最新下载
更多>
网站特效
网站源码
网站素材
前端模板
关于我们 免责声明 Sitemap
PHP中文网:公益在线PHP培训,帮助PHP学习者快速成长!