CLOVER🍀

That was when it all began.

Amazon SQSにメッセヌゞを送受信するアプリケヌションをSpring Bootで曞いおみるAWS SDK for Java v2、LocalStack利甚

Spring AWS LocalStack Terraform

これは、なにをしたくお曞いたもの

前に、Amazon SQSずいっおもLocalStack䞊のものですがを䜿っおFIFOキュヌを詊しおみたした。

LocalStackでAmazon SQSのFIFOキューを試してみる（AWS SDK for Javaを使用） - CLOVER🍀

今回は、垞駐アプリケヌションずしお曞いおみたらどうなるかなずいうこずで詊しおみたした。

お題

Spring Bootを䜿っお、以䞋のようなWebアプリケヌションずしお曞いおみたす。

  • AWS SDK for Java v2を䜿うSpring Cloud AWSは䜿甚しない
  • Amazon SQSに送信するメッセヌゞは、RestControllerで受け取っおキュヌに送信
  • Amazon SQSからメッセヌゞを受信する機胜は、アプリケヌションが停止するたで延々ずポヌリングする

今回は、特に埌者をポむントにしおいたす。
※ちなみに、受信のみを察象にする堎合はWebアプリケヌションにしなくおもOKです

䜜成するキュヌは、前回同様FIFOキュヌにしたす。

環境

今回の環境は、こちら。

$ java --version
openjdk 17.0.4 2022-07-19
OpenJDK Runtime Environment (build 17.0.4+8-Ubuntu-120.04)
OpenJDK 64-Bit Server VM (build 17.0.4+8-Ubuntu-120.04, mixed mode, sharing)


$ mvn --version
Apache Maven 3.8.6 (84538c9988a25aec085021c365c560670ad80f63)
Maven home: $HOME/.sdkman/candidates/maven/current
Java version: 17.0.4, vendor: Private Build, runtime: /usr/lib/jvm/java-17-openjdk-amd64
Default locale: ja_JP, platform encoding: UTF-8
OS name: "linux", version: "5.4.0-126-generic", arch: "amd64", family: "unix"

LocalStack。

$ python3 --version
Python 3.8.10


$ localstack --version
1.1.0

起動。

$ localstack start

Amazon SQSのキュヌの䜜成には、Terraformを䜿うこずにしたす。

$ terraform version
Terraform v1.3.1
on linux_amd64

Amazon SQSのキュヌを䜜成する

たずはAmazon SQSのキュヌを䜜成したす。

Terraformのリ゜ヌス定矩ファむルは、こんな感じで甚意。

main.tf

terraform {
  required_version = "1.3.1"

  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "4.33.0"
    }
  }
}

provider "aws" {
  access_key                  = "mock_access_key"
  region                      = "us-east-1"
  secret_key                  = "mock_secret_key"
  skip_credentials_validation = true
  skip_metadata_api_check     = true
  skip_requesting_account_id  = true

  endpoints {
    sqs = "http://localhost:4566"
  }
}

resource "aws_sqs_queue" "queue" {
  name                        = "my-queue.fifo"
  fifo_queue                  = true
  content_based_deduplication = true
}

output "queue_url" {
  value = aws_sqs_queue.queue.url
}

initしおapply。

$ terraform init
$ terraform apply

キュヌが䜜成されたした。

Apply complete! Resources: 1 added, 0 changed, 0 destroyed.

Outputs:

queue_url = "http://localhost:4566/000000000000/my-queue.fifo"

このURLを芚えおおきたす。

Spring Bootアプリケヌションを䜜成する

では、アプリケヌションを䜜成したしょう。

たずはSpring Bootプロゞェクトを䜜成。䟝存関係は、webのみにしおいたす。

$ curl -s https://start.spring.io/starter.tgz \
  -d bootVersion=2.7.4 \
  -d javaVersion=17 \
  -d name=spring-boot-sdk-v2-sqs \
  -d groupId=org.littlewings \
  -d artifactId=spring-boot-sdk-v2-sqs \
  -d version=0.0.1-SNAPSHOT \
  -d packageName=org.littlewings.spring.aws \
  -d dependencies=web \
  -d baseDir=spring-boot-sdk-v2-sqs | tar zxvf -

プロゞェクト内に移動。

$ cd spring-boot-sdk-v2-sqs

生成された゜ヌスコヌドは、削陀しおおきたす。

$ rm src/main/java/org/littlewings/spring/aws/SpringBootSdkV2SqsApplication.java src/test/java/org/littlewings/spring/aws/SpringBootSdkV2SqsApplicationTests.java

Maven䟝存関係やプラグむンの蚭定など。

        <properties>
                <java.version>17</java.version>
        </properties>
        <dependencies>
                <dependency>
                        <groupId>org.springframework.boot</groupId>
                        <artifactId>spring-boot-starter-web</artifactId>
                </dependency>

                <dependency>
                        <groupId>org.springframework.boot</groupId>
                        <artifactId>spring-boot-starter-test</artifactId>
                        <scope>test</scope>
                </dependency>
        </dependencies>

        <build>
                <plugins>
                        <plugin>
                                <groupId>org.springframework.boot</groupId>
                                <artifactId>spring-boot-maven-plugin</artifactId>
                        </plugin>
                </plugins>
        </build>

AWS SDK for Java v2を䜿うので、䟝存関係はこのように修正したした。

        <dependencyManagement>
                <dependencies>
                        <dependency>
                                <groupId>software.amazon.awssdk</groupId>
                                <artifactId>bom</artifactId>
                                <version>2.17.285</version>
                                <type>pom</type>
                                <scope>import</scope>
                        </dependency>
                </dependencies>
        </dependencyManagement>

        <dependencies>
                <dependency>
                        <groupId>org.springframework.boot</groupId>
                        <artifactId>spring-boot-starter-web</artifactId>
                </dependency>
                <dependency>
                        <groupId>software.amazon.awssdk</groupId>
                        <artifactId>sqs</artifactId>
                </dependency>

                <dependency>
                        <groupId>org.springframework.boot</groupId>
                        <artifactId>spring-boot-starter-test</artifactId>
                        <scope>test</scope>
                </dependency>
        </dependencies>

Apache Maven プロゞェクトの蚭定 / SDK を䟝存関係ずしお宣蚀する

SqsClientはメッセヌゞの送信でも受信でも䜿うので、Beanずしお定矩するこずにしたした。

src/main/java/org/littlewings/spring/aws/AwsConfig.java

package org.littlewings.spring.aws;

import java.net.URI;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
import software.amazon.awssdk.auth.credentials.AwsCredentials;
import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
import software.amazon.awssdk.awscore.defaultsmode.DefaultsMode;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.sqs.SqsClient;

@Configuration
public class AwsConfig {
    @Bean
    public SqsClient sqsClient() {
        AwsCredentials awsCredentials =
                AwsBasicCredentials.create("mock_access_key", "mock_secret_key");

        SqsClient sqsClient =
                SqsClient
                        .builder()
                        .defaultsMode(DefaultsMode.AUTO)
                        .credentialsProvider(StaticCredentialsProvider.create(awsCredentials))
                        .region(Region.US_EAST_1)
                        .endpointOverride(URI.create("http://localhost:4566"))
                        .build();

        return sqsClient;
    }
}

゚ンドポむントはLocalStackのものを蚭定し、Smart Configuration Defaultsはautoを遞択。

AWS SDK for Java v2のSmart Configuration Defaultsの紹介 | Amazon Web Services ブログ

Amazon SQSキュヌぞ、メッセヌゞの送信を行うRestController。

src/main/java/org/littlewings/spring/aws/SendMessageController.java

package org.littlewings.spring.aws;

import java.util.Map;
import java.util.UUID;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;
import software.amazon.awssdk.services.sqs.SqsClient;
import software.amazon.awssdk.services.sqs.model.SendMessageRequest;
import software.amazon.awssdk.services.sqs.model.SendMessageResponse;

@RestController
public class SendMessageController {
    SqsClient sqsClient;

    ObjectMapper objectMapper;

    public SendMessageController(SqsClient sqsClient, ObjectMapper objectMapper) {
        this.sqsClient = sqsClient;
        this.objectMapper = objectMapper;
    }

    @PostMapping(value = "send", consumes = "application/json")
    public String send(@RequestBody Map<String, Object> message) throws JsonProcessingException {
        String messageAsString = objectMapper.writeValueAsString(message);

        SendMessageRequest request =
                SendMessageRequest
                        .builder()
                        .queueUrl("http://localhost:4566/000000000000/my-queue.fifo")
                        .messageDeduplicationId(UUID.randomUUID().toString())
                        .messageGroupId("group-1")
                        .messageBody(messageAsString)
                        .build();

        SendMessageResponse response = sqsClient.sendMessage(request);

        return String.format(
                "send message = [%s], message-id = [%s], sequence-number = [%s]",
                messageAsString,
                response.messageId(),
                response.sequenceNumber()
        );
    }
}

リク゚ストボディをそのたたJSON文字列に倉換しお、メッセヌゞのボディずしおキュヌぞ送信したす。

参考にしおいるのは、このあたりです。

メッセヌゞの受信偎。

src/main/java/org/littlewings/spring/aws/MessageReceiver.java

package org.littlewings.spring.aws;

import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.springframework.core.task.TaskExecutor;
import org.springframework.stereotype.Component;
import software.amazon.awssdk.services.sqs.SqsClient;
import software.amazon.awssdk.services.sqs.model.DeleteMessageRequest;
import software.amazon.awssdk.services.sqs.model.DeleteMessageResponse;
import software.amazon.awssdk.services.sqs.model.ReceiveMessageRequest;
import software.amazon.awssdk.services.sqs.model.ReceiveMessageResponse;

@Component
public class MessageReceiver {
    SqsClient sqsClient;

    TaskExecutor taskExecutor;

    AtomicBoolean run = new AtomicBoolean(true);
    CountDownLatch latch = new CountDownLatch(1);

    public MessageReceiver(SqsClient sqsClient, TaskExecutor taskExecutor) {
        this.sqsClient = sqsClient;
        this.taskExecutor = taskExecutor;
    }

    @PostConstruct
    public void receiveMessageWhile() {
        taskExecutor.execute(() -> {
            while (run.get()) {
                ReceiveMessageRequest request =
                        ReceiveMessageRequest
                                .builder()
                                .queueUrl("http://localhost:4566/000000000000/my-queue.fifo")
                                .receiveRequestAttemptId(UUID.randomUUID().toString())
                                .maxNumberOfMessages(3)
                                .waitTimeSeconds(5)
                                .build();

                ReceiveMessageResponse response = sqsClient.receiveMessage(request);

                response.messages().forEach(message -> {
                    System.out.printf(
                            "receive message = [%s], message-id = [%s], receipt-handle=[%s]%n",
                            message.body(),
                            message.messageId(),
                            message.receiptHandle()
                    );

                    DeleteMessageRequest deleteRequest =
                            DeleteMessageRequest
                                    .builder()
                                    .queueUrl("http://localhost:4566/000000000000/my-queue.fifo")
                                    .receiptHandle(message.receiptHandle())
                                    .build();

                    DeleteMessageResponse deleteResponse = sqsClient.deleteMessage(deleteRequest);
                    if (deleteResponse.sdkHttpResponse().isSuccessful()) {
                        System.out.printf("message[%s] deleted%n", message.messageId());
                    } else {
                        System.out.printf(
                                "delete message failure, code = [%d], text = [%s]%n",
                                deleteResponse.sdkHttpResponse().statusCode(),
                                deleteResponse.sdkHttpResponse().statusText()
                        );
                    }
                });

                if (run.get()) {
                    System.out.println("waiting, next messages...");
                } else {
                    System.out.println("enter shutdown sequence");
                    latch.countDown();
                }
            }
        });
    }

    @PreDestroy
    public void shutdown() throws InterruptedException {
        run.set(false);

        latch.await();
        System.out.println("shutdown");
    }
}

こちらは、@Componentずしお定矩。

@Component
public class MessageReceiver {

メッセヌゞの受信は、@PostConstructを䜿っおBeanが構築され次第開始するようにしおいたす。

    @PostConstruct
    public void receiveMessageWhile() {
        taskExecutor.execute(() -> {
            while (run.get()) {
                ReceiveMessageRequest request =
                        ReceiveMessageRequest
                                .builder()
                                .queueUrl("http://localhost:4566/000000000000/my-queue.fifo")
                                .receiveRequestAttemptId(UUID.randomUUID().toString())
                                .maxNumberOfMessages(3)
                                .waitTimeSeconds(5)
                                .build();

                ReceiveMessageResponse response = sqsClient.receiveMessage(request);

                response.messages().forEach(message -> {
                    System.out.printf(
                            "receive message = [%s], message-id = [%s], receipt-handle=[%s]%n",
                            message.body(),
                            message.messageId(),
                            message.receiptHandle()
                    );

                    DeleteMessageRequest deleteRequest =
                            DeleteMessageRequest
                                    .builder()
                                    .queueUrl("http://localhost:4566/000000000000/my-queue.fifo")
                                    .receiptHandle(message.receiptHandle())
                                    .build();

                    DeleteMessageResponse deleteResponse = sqsClient.deleteMessage(deleteRequest);
                    if (deleteResponse.sdkHttpResponse().isSuccessful()) {
                        System.out.printf("message[%s] deleted%n", message.messageId());
                    } else {
                        System.out.printf(
                                "delete message failure, code = [%d], text = [%s]%n",
                                deleteResponse.sdkHttpResponse().statusCode(),
                                deleteResponse.sdkHttpResponse().statusText()
                        );
                    }
                });

                if (run.get()) {
                    System.out.println("waiting, next messages...");
                } else {
                    System.out.println("enter shutdown sequence");
                    latch.countDown();
                }
            }
        });
    }

アプリケヌションが起動しおいる間は、メッセヌゞの受信を継続するように無限ルヌプにしおいたす。

            while (run.get()) {
                ReceiveMessageRequest request =
                        ReceiveMessageRequest
                                .builder()
                                .queueUrl("http://localhost:4566/000000000000/my-queue.fifo")
                                .receiveRequestAttemptId(UUID.randomUUID().toString())
                                .maxNumberOfMessages(3)
                                .waitTimeSeconds(5)
                                .build();

                ReceiveMessageResponse response = sqsClient.receiveMessage(request);

この凊理をそのたた曞いおしたうずスレッドを専有しおアプリケヌションが起動しなくなっおしたうので、TaskExecutorを䜿っお
別スレッドで実行するようにしたした。

        taskExecutor.execute(() -> {
            while (run.get()) {

Integration / Task Execution and Scheduling / The Spring TaskExecutor Abstraction / Using a TaskExecutor

停止に関しおは、受信したメッセヌゞを凊理しおいる最䞭に止たっお欲しくはありたせん。

そこで、今回はCoundDownLatchを䜿いたした。@PreDestroyでwhileの条件になっおいるAtomicBooleanの倉数をfalseにしお、
CountDownLatchがカりントダりンするのを埅ちたす。

    @PreDestroy
    public void shutdown() throws InterruptedException {
        run.set(false);

        latch.await();
        System.out.println("shutdown");
    }

カりントダりンは、先ほどのメッセヌゞ受信のルヌプ内で行われたす。

                if (run.get()) {
                    System.out.println("waiting, next messages...");
                } else {
                    System.out.println("enter shutdown sequence");
                    latch.countDown();
                }

そしおCountDownLatchの倀は1で蚭定しおいるので、1回カりントダりンすればブロックが解陀されたす。

    CountDownLatch latch = new CountDownLatch(1);

これで、凊理しおいるメッセヌゞがない状態になれば停止したす。

メッセヌゞの受信は、受信したメッセヌゞを暙準出力に曞き出したら、メッセヌゞを削陀する流れで実装しおいたす。

                ReceiveMessageRequest request =
                        ReceiveMessageRequest
                                .builder()
                                .queueUrl("http://localhost:4566/000000000000/my-queue.fifo")
                                .receiveRequestAttemptId(UUID.randomUUID().toString())
                                .maxNumberOfMessages(3)
                                .waitTimeSeconds(5)
                                .build();

                ReceiveMessageResponse response = sqsClient.receiveMessage(request);

                response.messages().forEach(message -> {
                    System.out.printf(
                            "receive message = [%s], message-id = [%s], receipt-handle=[%s]%n",
                            message.body(),
                            message.messageId(),
                            message.receiptHandle()
                    );

                    DeleteMessageRequest deleteRequest =
                            DeleteMessageRequest
                                    .builder()
                                    .queueUrl("http://localhost:4566/000000000000/my-queue.fifo")
                                    .receiptHandle(message.receiptHandle())
                                    .build();

                    DeleteMessageResponse deleteResponse = sqsClient.deleteMessage(deleteRequest);
                    if (deleteResponse.sdkHttpResponse().isSuccessful()) {
                        System.out.printf("message[%s] deleted%n", message.messageId());
                    } else {
                        System.out.printf(
                                "delete message failure, code = [%d], text = [%s]%n",
                                deleteResponse.sdkHttpResponse().statusCode(),
                                deleteResponse.sdkHttpResponse().statusText()
                        );
                    }

やっぱり、このあたりを参考に。

メッセヌゞの受信はロングポヌリングで行っおいお、waitTimeSecondsを5に蚭定しおいるので受信時に最倧で5秒ブロックしたす。

                ReceiveMessageRequest request =
                        ReceiveMessageRequest
                                .builder()
                                .queueUrl("http://localhost:4566/000000000000/my-queue.fifo")
                                .receiveRequestAttemptId(UUID.randomUUID().toString())
                                .maxNumberOfMessages(3)
                                .waitTimeSeconds(5)
                                .build();

Amazon SQS ショヌトポヌリングずロングポヌリング / ロングポヌリングを䜿甚したメッセヌゞの消費

5秒経っおもメッセヌゞがこなかったら諊めお次のルヌプぞ移り、埅機䞭にメッセヌゞが来たらすぐに凊理を行っお次のルヌプぞ移りたす。

mainメ゜ッドを持ったクラス。

src/main/java/org/littlewings/spring/aws/App.java

package org.littlewings.spring.aws;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableAsync;

@SpringBootApplication
public class App {
    public static void main(String... args) {
        SpringApplication.run(App.class, args);
    }
}

アプリケヌションの蚭定。なんずなく、Graceful Shutdownも有効にしおおきたした。
※蚭定しなくおも、今回の動䜜確認は可胜です

src/main/resources/application.properties

server.shutdown=graceful
spring.lifecycle.timeout-per-shutdown-phase=15s

確認する

では、動かしお確認しおみたしょう。

パッケヌゞングしお

$ mvn package

実行。

$ java -jar target/spring-boot-sdk-v2-sqs-0.0.1-SNAPSHOT.jar

アプリケヌションが起動したら、5秒おきに暙準出力にメッセヌゞが流れたす。

waiting, next messages...
waiting, next messages...

メッセヌゞを送っおみたしょう。

$ curl -XPOST -H 'Content-Type: application/json' localhost:8080/send -d '{"messsage": "Hello World"}'
send message = [{"messsage":"Hello World"}], message-id = [973ccd24-0fc6-4996-9d0a-78623c13eb15], sequence-number = [14300583401512501286]

メッセヌゞIDやシヌケンス番号も取埗できおいたす。

アプリケヌション偎の暙準出力には、こんな内容が出力されたす。

receive message = [{"messsage":"Hello World"}], message-id = [973ccd24-0fc6-4996-9d0a-78623c13eb15], receipt-handle=[ZjNiYjI3NTQtYzg1OS00YWRlLThlOTEtNDMwMmJlMWJiMzIxIGFybjphd3M6c3FzOnVzLWVhc3QtMTowMDAwMDAwMDAwMDA6bXktcXVldWUuZmlmbyA5NzNjY2QyNC0wZmM2LTQ5OTYtOWQwYS03ODYyM2MxM2ViMTUgMTY2NDgxNDE1My45NzAwOTk=]
message[973ccd24-0fc6-4996-9d0a-78623c13eb15] deleted

そしお、次のルヌプぞ移りたす。

waiting, next messages...
waiting, next messages...

この埌にメッセヌゞを送り続けおみたす。

$ curl -XPOST -H 'Content-Type: application/json' localhost:8080/send -d '{"messsage": "Hello World"}'
send message = [{"messsage":"Hello World"}], message-id = [581814e9-98d6-48f4-8878-24877f81e2eb], sequence-number = [14300583401512501287]


$ curl -XPOST -H 'Content-Type: application/json' localhost:8080/send -d '{"messsage": "Hello World"}'
send message = [{"messsage":"Hello World"}], message-id = [595dc1ff-0736-4efa-9c1b-80d5a5dd2214], sequence-number = [14300583401512501288]


$ curl -XPOST -H 'Content-Type: application/json' localhost:8080/send -d '{"messsage": "Hello World"}'
send message = [{"messsage":"Hello World"}], message-id = [a9a07acb-071e-46ce-bd6e-194ac33bf048], sequence-number = [14300583401512501289]

アプリケヌション偎は、メッセヌゞの受信を続けたす。

receive message = [{"messsage":"Hello World"}], message-id = [581814e9-98d6-48f4-8878-24877f81e2eb], receipt-handle=[OTgzODRmODItMTgyYy00YTRjLWIyZmEtMjY4YTMxZjM5YmFjIGFybjphd3M6c3FzOnVzLWVhc3QtMTowMDAwMDAwMDAwMDA6bXktcXVldWUuZmlmbyA1ODE4MTRlOS05OGQ2LTQ4ZjQtODg3OC0yNDg3N2Y4MWUyZWIgMTY2NDgxNDI1MS41NDc4MTE=]
message[581814e9-98d6-48f4-8878-24877f81e2eb] deleted
waiting, next messages...
receive message = [{"messsage":"Hello World"}], message-id = [595dc1ff-0736-4efa-9c1b-80d5a5dd2214], receipt-handle=[NTQ4NDc1ZjEtMTE1ZC00ZTU3LWJmZTUtYjkyNTU2Y2IzOGI3IGFybjphd3M6c3FzOnVzLWVhc3QtMTowMDAwMDAwMDAwMDA6bXktcXVldWUuZmlmbyA1OTVkYzFmZi0wNzM2LTRlZmEtOWMxYi04MGQ1YTVkZDIyMTQgMTY2NDgxNDI1Mi4yNDUwMzQy]
message[595dc1ff-0736-4efa-9c1b-80d5a5dd2214] deleted
waiting, next messages...
receive message = [{"messsage":"Hello World"}], message-id = [a9a07acb-071e-46ce-bd6e-194ac33bf048], receipt-handle=[YTNmZGVkZjMtYWFkZC00MTkzLWE0MDAtZmI1ZjFjMGI2YjE4IGFybjphd3M6c3FzOnVzLWVhc3QtMTowMDAwMDAwMDAwMDA6bXktcXVldWUuZmlmbyBhOWEwN2FjYi0wNzFlLTQ2Y2UtYmQ2ZS0xOTRhYzMzYmYwNDggMTY2NDgxNDI1Mi43OTI3MDc=]
message[a9a07acb-071e-46ce-bd6e-194ac33bf048] deleted
waiting, next messages...

OKみたいですね。

最埌にCtrl-cでアプリケヌションを停止。

2022-10-04 01:25:27.552  INFO 48599 --- [ionShutdownHook] o.s.b.w.e.tomcat.GracefulShutdown        : Commencing graceful shutdown. Waiting for active requests to complete
2022-10-04 01:25:27.555  INFO 48599 --- [tomcat-shutdown] o.s.b.w.e.tomcat.GracefulShutdown        : Graceful shutdown complete
enter shutdown sequence
shutdown

Beanの砎棄時に、メッセヌゞの受信を停止するように動䜜しおいるこずが確認できたした。

OKですね。

オマケ

お題にも曞きたしたが、受信の郚分だけであればspring-boot-starter-webではなくspring-boot-starterでOKです。

     <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>

TaskExecutorを䜿っお凊理を実行したたたなので、これでもアプリケヌションは終了せずに実行され続けたす。

停止時の凊理が行われるのも同じです。

たずめ

Amazon SQSを䜿っお、メッセヌゞを送受信するプログラムを曞いおみたした。

今回はずっずメッセヌゞを埅ち続けるアプリケヌションにしたかったので、Webアプリケヌションにし぀぀、TaskExecutorを䜿っお
別スレッドでメッセヌゞの受信を行う構成にしたした。

なにも考えずに無限に受信凊理を行うず、そこでアプリケヌションがブロックしお先に進たなくなるなどのお決たりを螏んだりをしたしたが、
ずりあえず目暙は達成できたのでよしずしたしょう。

Infinispan 14.0のRESP゚ンドポむントをRedis CLIで詊す

Infinispan

これは、なにをしたくお曞いたもの

Infinispan 14.0で「RESP endpoint」ずいうものが実装され、Redis互換の゚ンドポむントをサポヌトするようになったずいうのでちょっず
詊しおみたした。

Infinispan 14.0.0.Final

なお、このブログには曞かれおいたせんが、珟時点でRESP゚ンドポむントは「実隓的モゞュヌル」扱いです。

RESP゚ンドポむント

RESP゚ンドポむントに関するドキュメントは、こちら。

Using the RESP protocol endpoint with Infinispan

以䞋のようなドキュメントの䞀芧ペヌゞには茉っおいないのですが、

Guides

Infinispan 14.0 documentation index

各ドキュメント内の「Document Index」の䞭から「RESP protocol endpoint」を遞択するず芋るこずができたす。

ドキュメントによるず、RESP゚ンドポむントはRESP3を実装した実隓的モゞュヌルのようです。RESP゚ンドポむントを䜿うず、
Infinispan Serverをバック゚ンドずするRESPサヌバヌに察しおRedisクラむアントから接続し、Cache操䜜を行うこずができたす。

Infinispan Server includes an experimental module that implements the RESP3 protocol. The RESP endpoint allows Redis clients to connect to one or more one or several Infinispan-backed RESP servers and perform cache operations.

Using the RESP endpoint

RESP゚ンドポむントのモゞュヌルは、こちら。

https://github.com/infinispan/infinispan/tree/14.0.0.Final/server/resp

RESP3自䜓に぀いおは、RedisのGitHubリポゞトリに蚘茉がありたす。

https://github.com/redis/redis-specifications/blob/master/protocol/RESP3.md

RESPずいうのは、「Redis Serialization Protocol」の略のようです。

Redis Serialization Protocol (RESP)

https://github.com/redis/redis-specifications/blob/master/protocol/README.md

RESPずは、クラむアントずサヌバヌの間のリク゚スト、レスポンスをどのように扱うかを蚭蚈したプロトコルです。クラむアントがなんらかの
リク゚ストを実行し、サヌバヌがなんらかのデヌタを返すようなものです。

The protocol is designed to handle request-response chats between clients and servers, where the client performs some kind of request, and the server replies with some data.

RESP3 overview

RESP3では、以䞋のデヌタ型を扱えたす。

  • 配列
  • Blob
  • 文字列
  • ゚ラヌ
  • Number
  • Null
  • Double
  • Boolean
  • Blob゚ラヌ
  • Verbatim string
  • Map
  • Set
  • Attribute
  • Push
  • Hello

RESP3 types

Infinispan ServerのRESP3゚ンドポむントでサポヌトされおいるコマンドは、以䞋になりたす。

  • AUTH
  • DECR
  • DEL
  • ECHO
  • GET
  • HELLO
  • INCR
  • MGET
  • MSET
  • PING
  • PUBLISH
  • QUIT
  • RESET
  • SET
  • SUBSCRIBE
  • UNSUBSCRIBE

Using the RESP protocol endpoint with Infinispan / Redis commands

Redis本家ず比べるず、ただただ少ないですね。

Commands | Redis

InfinispanずRedis

RESP゚ンドポむントに察するチケットはこちらです。

[ISPN-12439] Add server endpoint that implements RESP protocol from Redis - Red Hat Issue Tracker

このチケットにはInfinispan ServerにRESPを䜿っおRedisクラむアントから接続できれば䟿利だ、ず曞かれおいるのですが。

It could be beneficial to have a server endpoint that we can enable to allow a Redis client to connect to Infinispan Server using the RESP protocol to perform cache operations.

Redisずしお䜿えた方が、Infinispan Serverのナヌスケヌスが増えるんですかね 
Redisの代替ずしお䜿いたいずいうリク゚ストが倚かったのでしょうか

個人的には「今、この機胜を远加するんだ 」ず思ったので、もう少し背景が知りたかったですね。

それはさおおき、ちょっず詊しおみるずしたしょう。

環境

今回の環境は、こちら。

$ java --version
openjdk 17.0.4.1 2022-08-12
OpenJDK Runtime Environment Temurin-17.0.4.1+1 (build 17.0.4.1+1)
OpenJDK 64-Bit Server VM Temurin-17.0.4.1+1 (build 17.0.4.1+1, mixed mode, sharing)


$ bin/server.sh --version

Infinispan Server 14.0.0.Final (Flying Saucer)
Copyright (C) Red Hat Inc. and/or its affiliates and other contributors
License Apache License, v. 2.0. http://www.apache.org/licenses/LICENSE-2.0

Infinispan Serverは、172.17.0.2で動䜜しおいるものずしたす。

起動は、以䞋のコマンドで行っおおきたす。

$ bin/server.sh \
    -b 0.0.0.0 \
    -Djgroups.tcp.address=`hostname -i`

RedisはCLIのみ䜿いたす。

$ bin/redis-cli --version
redis-cli 7.0.5

RESP゚ンドポむントにRedis CLIから接続する

では、RESP゚ンドポむントにRedis CLIから接続しおみたしょう。

なお、Infinispan Serverを起動するず気づきたすが、RESP゚ンドポむントはデフォルトで有効になっおいるようです。

2022-10-02 14:44:12,868 INFO  (ForkJoinPool.commonPool-worker-1) [org.infinispan.SERVER] ISPN080018: Started connector Resp (internal)

デフォルトの状態では、認蚌情報なしで接続しようずするず゚ラヌになりたす。

$ bin/redis-cli -h 172.17.0.2 -p 11222
172.17.0.2:11222> set key1 value1
(error) WRONGPASS invalid username-password pair or user is disabled.

そこで、たずはInfinispan Serverにナヌザヌを远加したす。今回は、管理ナヌザヌずアプリケヌションナヌザヌを远加したした。

$ bin/cli.sh user create -g admin -p password ispn-admin
$ bin/cli.sh user create -g application -p password ispn-user

アプリケヌションナヌザヌのナヌザヌ名、パスワヌドを䜿っお、もう1床Redis CLIから接続。

$ bin/redis-cli -h 172.17.0.2 -p 11222 --user ispn-user --pass password
Warning: Using a password with '-a' or '-u' option on the command line interface may not be safe.

今床は接続できたので、setおよびgetコマンドで確認。

172.17.0.2:11222> set key1 value1
OK

172.17.0.2:11222> get key1
"value1"

すごく、あっさり動きたした 。

このデヌタが氞続化されるかどうかなどは、RedisのふりをしおいるInfinispan ServerのCacheの蚭定次第ずいうこずになるわけですが。

RESP゚ンドポむントが䜿甚するCacheに぀いお

こうあっさり動いおしたうず、いろいろ気になりたす。たず、デヌタはどのCacheに保存されおいるのでしょうか

Infinispanの管理CLIでログむン。

$ bin/cli.sh -c-
Username: ispn-admin
Password:

Cacheを芋るず、respCacheずいう芋慣れないCacheがありたす。

[c6b2176b23ba-18766@cluster//containers/default]> ls caches
___script_cache
respCache

デヌタを芋おみたす。

[c6b2176b23ba-18766@cluster//containers/default]> ls caches/respCache
Key
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
key1


[c6b2176b23ba-18766@cluster//containers/default]> cache respCache
[c6b2176b23ba-18766@cluster//containers/default]> get key1
value1

確かに、REPS゚ンドポむントで䜿われおいるCacheのようです。

Cacheの皮類は、Replicated Cacheでした。

describe caches/respCache
{
  "respCache" : {
    "replicated-cache" : {
      "mode" : "SYNC",
      "encoding" : {
        "key" : {
          "media-type" : "text/plain"
        },
        "value" : {
          "media-type" : "application/octet-stream"
        }
      }
    }
  }
}

このCache名は、RESP゚ンドポむントのドキュメントには蚘茉がありたせんでしたが、XML Schemaの方には蚘茉がありたした。

Names the cache that the RESP connector exposes. The default cache name is respCache.

urn:infinispan:server:14.0

respCacheずいうのは、RESP゚ンドポむントが䜿うデフォルトのCache名のようです。

https://github.com/infinispan/infinispan/blob/14.0.0.Final/server/resp/src/main/java/org/infinispan/server/resp/configuration/RespServerConfiguration.java#L23

このReplicated Cacheは、起動時に䜜成するようです。

https://github.com/infinispan/infinispan/blob/14.0.0.Final/server/resp/src/main/java/org/infinispan/server/resp/RespServer.java#L43-L56

そしお、クラスタリングが有効になっおいる堎合はReplicated Cacheに蚭定されたす。

            if (cacheManager.getCacheManagerConfiguration().isClustered()) { // We are running in clustered mode
               builder.clustering().cacheMode(CacheMode.REPL_SYNC);
            }

メディアタむプは、明瀺的に指定されおいたすね。キヌはtext/plain、倀はapplication/octet-streamですね。

            builder.encoding().key().mediaType(MediaType.TEXT_PLAIN_TYPE);
            builder.encoding().value().mediaType(MediaType.APPLICATION_OCTET_STREAM_TYPE);

RESP゚ンドポむントの蚭定を倉えおみる

ドキュメントを芋おいるず、resp-connectorのcache属性で䜿甚するCacheを蚭定できそうです。

Using the RESP protocol endpoint with Infinispan / Enabling the RESP endpoint

ずいっおも、起動時にCacheを䜜成しおしたうのでした。

実際、デフォルトのinfinispan.xmlにはRESP゚ンドポむント甚のCacheの定矩はありたせん。

   <cache-container name="default" statistics="true">
      <transport cluster="${infinispan.cluster.name:cluster}" stack="${infinispan.cluster.stack:tcp}" node-name="${infinispan.node.name:}"/>
      <security>
         <authorization/>
      </security>
   </cache-container>

なんなら、゚ンドポむントの定矩もなにもないわけですが。

      <endpoints socket-binding="default" security-realm="default" />

ずいうこずは、Cacheを明瀺的に定矩する堎合はInfinispan Serverの起動前に行う必芁がありたすね。

たずは、ドキュメントに習っおRESP゚ンドポむントを蚭定しおみたす。myRespCacheずいうCacheを䜿うように蚭定しおみたした。

      <endpoints socket-binding="default" security-realm="default">
        <endpoint>
          <resp-connector cache="myRespCache">
            <authentication/>
          </resp-connector>
          <hotrod-connector>
            <authentication>
              <sasl mechanisms="SCRAM-SHA-512 SCRAM-SHA-384 SCRAM-SHA-256
                                SCRAM-SHA-1 DIGEST-SHA-512 DIGEST-SHA-384
                                DIGEST-SHA-256 DIGEST-SHA DIGEST-MD5 PLAIN"
                    server-name="infinispan"
                    qop="auth"/>
            </authentication>
          </hotrod-connector>
          <rest-connector>
            <authentication mechanisms="DIGEST BASIC"/>
          </rest-connector>
        </endpoint>
      </endpoints>

Hot Rod゚ンドポむントずREST゚ンドポむントは明瀺的に蚭定が必芁です。

認蚌が有効になっおいるInfinispan Serverでは、Hot Rod゚ンドポむントたたはREST゚ンドポむントの少なくずもどちらかは認蚌の蚭定が
行われおいる必芁があるようです。

https://github.com/infinispan/infinispan/blob/14.0.0.Final/server/runtime/src/main/java/org/infinispan/server/configuration/endpoint/EndpointsConfigurationBuilder.java#L77-L86

単にRESP゚ンドポむントの蚭定だけすれば良いず思っお、以䞋のような状態にしおしたうず

      <endpoints socket-binding="default" security-realm="default">
        <endpoint>
          <resp-connector cache="myRespCache">
            <authentication/>
          </resp-connector>
        </endpoint>
      </endpoints>

Infinispan Serverが起動に倱敗したす。

2022-10-02 16:09:28,754 FATAL (main) [org.infinispan.SERVER] ISPN080028: Infinispan Server failed to start org.infinispan.commons.CacheConfigurationException: ISPN080070: The cache container requires authorization, but none of the connectors enable authentication

endpointを定矩したら、Hot Rod゚ンドポむントもREST゚ンドポむントも明瀺的に蚭定しなければいけたせん。

気を取り盎しお、Infinispan Serverを再起動したす。

するず、RESP゚ンドポむントに指定した名前でCacheが䜜成されおいたした。

Redis CLIでデヌタを保存したりするずこのCacheに栌玍されたすが、確認は省略。

次に、infinispan.xmlに明瀺的にCacheを定矩しお、そちらを䜿うようにしおみたしょう。

今回はDistributed Cacheを定矩。

   <cache-container name="default" statistics="true">
      <transport cluster="${infinispan.cluster.name:cluster}" stack="${infinispan.cluster.stack:tcp}" node-name="${infinispan.node.name:}"/>

      <distributed-cache name="distributedRespCache">
        <encoding>
          <key media-type="text/plain"/>
          <value media-type="application/octet-stream"/>
        </encoding>
      </distributed-cache>

      <security>
         <authorization/>
      </security>
   </cache-container>

RESP゚ンドポむントは、distributedRespCacheを参照するように蚭定。

      <endpoints socket-binding="default" security-realm="default">
        <endpoint>
          <resp-connector cache="distributedRespCache">
            <authentication/>
          </resp-connector>
          <hotrod-connector>
            <authentication>
              <sasl mechanisms="SCRAM-SHA-512 SCRAM-SHA-384 SCRAM-SHA-256
                                SCRAM-SHA-1 DIGEST-SHA-512 DIGEST-SHA-384
                                DIGEST-SHA-256 DIGEST-SHA DIGEST-MD5 PLAIN"
                    server-name="infinispan"
                    qop="auth"/>
            </authentication>
          </hotrod-connector>
          <rest-connector>
            <authentication mechanisms="DIGEST BASIC"/>
          </rest-connector>
        </endpoint>
      </endpoints>

これで、Infinispan Serverを再起動。

定矩したDistributed Cacheが珟れたす。

[3f666276f239-57422@cluster//containers/default]> ls caches
distributedRespCache
___script_cache

定矩の確認。

[3f666276f239-57422@cluster//containers/default]> describe caches/distributedRespCache
{
  "distributedRespCache" : {
    "distributed-cache" : {
      "mode" : "SYNC",
      "remote-timeout" : "17500",
      "statistics" : true,
      "encoding" : {
        "key" : {
          "media-type" : "text/plain"
        },
        "value" : {
          "media-type" : "application/octet-stream"
        }
      },
      "locking" : {
        "concurrency-level" : "1000",
        "acquire-timeout" : "15000",
        "striping" : false
      },
      "state-transfer" : {
        "timeout" : "60000"
      }
    }
  }
}

Redis CLIから、再床デヌタを保存しおみたしょう。

172.17.0.2:11222> set key1 value1
OK
172.17.0.2:11222> get key1
"value1"

Infinispan偎からも確認しおみたす。

[3f666276f239-57422@cluster//containers/default]> ls caches/distributedRespCache
Key
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
key1
[3f666276f239-57422@cluster//containers/default]> cache distributedRespCache
[3f666276f239-57422@cluster//containers/default/caches/distributedRespCache]> get key1
value1

OKですね。

オマケ

RESP゚ンドポむントで実装されおいるコマンドを確認するは、このあたりを芋るずよさそうです。

https://github.com/infinispan/infinispan/blob/14.0.0.Final/server/resp/src/main/java/org/infinispan/server/resp/Resp3AuthHandler.java

https://github.com/infinispan/infinispan/blob/14.0.0.Final/server/resp/src/main/java/org/infinispan/server/resp/Resp3Handler.java

https://github.com/infinispan/infinispan/blob/14.0.0.Final/server/resp/src/main/java/org/infinispan/server/resp/SubscriberHandler.java

たた、内郚的にはLettuceも䜿われおいたりしお、ちょっず面癜いです。

https://github.com/infinispan/infinispan/blob/14.0.0.Final/server/resp/src/main/java/org/infinispan/server/resp/RespLettuceHandler.java

たずめ

Infinispan 14.0で远加された、RESP゚ンドポむントを詊しおみたした。

割ず簡単に䜿えるのですが、ちょっず動きを倉えようずするず癖がある ずいうかドキュメントがあたり曞かれおいない感じがするので
ちょっず泚意ですね。

個人的には、この機胜の䜿いどころが気になるずころですが 実隓的モゞュヌルのようなので、もう少し様子をみたしょうか。