Съобщения в облака на платформата Red Hat OpenShift с помощта на Quarkus и AMQ Online

Здравейте всички! Ето го и последният ни пост от поредицата Quarkus! (Между другото, гледайте нашия уебинар „Това е Quarkus – родната Java рамка на Kubernetes“. Ще ви покажем как да започнете от нулата или да прехвърлите готови решения)

В предишен В тази публикация разгледахме съответните инструменти, които могат да се използват за количествено определяне на подобренията, получени в резултат на модернизиране на Java приложения.

От версия 0.17.0, Кваркус поддържа използването на Advanced Message Queuing Protocol (AMQP), който е отворен стандарт за прехвърляне на бизнес съобщения между приложения или организации.

Red Hat AMQ онлайн е услуга, изградена на базата на проект с отворен код EnMasse и внедряване на базиран на платформа механизъм за съобщения Red Hat OpenShift. За повече подробности как работи вижте тук (EN). Днес ще ви покажем как да комбинирате AMQ Online и Quarkus, за да изградите модерна система за съобщения, базирана на OpenShift, като използвате две нови технологии за съобщения.

Предполага се, че вече сте разположили AMQ Online на платформата OpenShift (ако не, вижте ръководство за инсталиране).

Като начало ще създадем приложение Quarkus, което ще бъде проста система за обработка на поръчки, използваща реактивни съобщения. Това приложение ще включва генератор на поръчки, който изпраща поръчки към опашка от съобщения на фиксиран интервал, както и процесор за обработка на поръчки, който ще обработва съобщения от опашката и ще генерира потвърждения, видими в браузъра.

След като създадем приложението, ще ви покажем как да вградите конфигурацията на системата за съобщения в приложението и да използвате AMQ Online, за да предоставим ресурсите, от които се нуждаем в системата.

Приложение Quarkus

Нашето приложение Quarkus работи на OpenShift и е модифицирана версия на програмата amqp-бърз старт. Можете да намерите пълен пример от страна на клиента тук.

Генератор на поръчки

Генераторът просто монотонно изпраща нарастващи идентификатори на поръчки до адреса на „поръчки“ на всеки 5 секунди.

@ApplicationScoped
public class OrderGenerator {
 
    private int orderId = 1;
 
    @Outgoing("orders")
    public Flowable<Integer> generate() {
        return Flowable.interval(5, TimeUnit.SECONDS)
        .map(tick -> orderId++);
    }
}

Процесор на поръчки

Манипулаторът на поръчките е още по-прост, той просто връща идентификатор за потвърждение на адреса за „потвърждения“.

@ApplicationScoped
public class OrderProcessor {
    @Incoming("orders")
    @Outgoing("confirmations")
    public Integer process(Integer order) {
        // Идентификатор подтверждения равен удвоенному идентификатору заказа <img draggable="false" class="emoji" alt=":-)" src="https://s.w.org/images/core/emoji/11.2.0/svg/1f642.svg">
        return order * 2;
    }
}

Ресурси за потвърждение

Ресурсът за потвърждение е HTTP крайна точка за изброяване на потвържденията, генерирани от нашето приложение.

@Path("/confirmations")
public class ConfirmationResource {
 
    @Inject
    @Stream("confirmations") Publisher<Integer> orders;
 
    @GET
    @Produces(MediaType.TEXT_PLAIN)
    public String hello() {
        return "hello";
    }
 
 
    @GET
    @Path("/stream")
    @Produces(MediaType.SERVER_SENT_EVENTS)
    public Publisher<Integer> stream() {
        return orders;
    }
}

регулиране

За да се свърже с AMQ Online, нашето приложение ще се нуждае от някои конфигурационни данни, а именно: конфигурация на конектор Quarkus, информация за крайната точка на AMQP и идентификационни данни на клиента. Разбира се, по-добре е всички конфигурационни данни да се съхраняват на едно място, но ние умишлено ще ги разделим, за да покажем възможните опции за конфигуриране на приложението Quarkus.

Съединители

Конфигурацията на конектора може да бъде предоставена по време на компилиране с помощта на файл със свойства на приложението:

mp.messaging.outgoing.orders.connector=smallrye-amqp
mp.messaging.incoming.orders.connector=smallrye-amqp

За да опростим нещата, ще използваме опашка от съобщения само за адреса на „поръчки“. А адресът за „потвърждения“ в нашето приложение ще използва опашка в паметта.

Крайна точка на AMQP

По време на компилиране името на хоста и номерът на порта за крайната точка на AMQP са неизвестни, така че трябва да бъдат инжектирани. Крайната точка може да бъде зададена в конфигурационната карта, която е създадена от AMQ Online, така че ще ги дефинираме чрез променливи на средата в манифеста на приложението:

spec:
  template:
    spec:
      containers:
      - env:
        - name: AMQP_HOST
          valueFrom:
            configMapKeyRef:
              name: quarkus-config
              key: service.host
        - name: AMQP_PORT
          valueFrom:
            configMapKeyRef:
              name: quarkus-config
              key: service.port.amqp

Акредитивни писма

Токенът на акаунта на услугата може да се използва за удостоверяване на нашето приложение в OpenShift. За да направите това, първо трябва да създадете персонализиран ConfigSource, който ще прочете токена за удостоверяване от файловата система на pod:

public class MessagingCredentialsConfigSource implements ConfigSource {
    private static final Set<String> propertyNames;
 
    static {
        propertyNames = new HashSet<>();
        propertyNames.add("amqp-username");
        propertyNames.add("amqp-password");
    }
 
    @Override
    public Set<String> getPropertyNames() {
        return propertyNames;
    }
 
    @Override
    public Map<String, String> getProperties() {
        try {
            Map<String, String> properties = new HashMap<>();
            properties.put("amqp-username", "@@serviceaccount@@");
            properties.put("amqp-password", readTokenFromFile());
            return properties;
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
    }
 
    @Override
    public String getValue(String key) {
        if ("amqp-username".equals(key)) {
            return "@@serviceaccount@@";
        }
        if ("amqp-password".equals(key)) {
            try {
                return readTokenFromFile();
            } catch (IOException e) {
                throw new UncheckedIOException(e);
            }
        }
        return null;
    }
 
    @Override
    public String getName() {
        return "messaging-credentials-config";
    }
 
    private static String readTokenFromFile() throws IOException {
        return new String(Files.readAllBytes(Paths.get("/var/run/secrets/kubernetes.io/serviceaccount/token")), StandardCharsets.UTF_8);
    }
}

Изградете и разположете приложението

Тъй като приложението трябва да бъде компилирано в изпълним файл, е необходима виртуална машина GraalVM. За подробности как да настроите среда за това вижте съответните инструкции в Ръководство за Quarkus.

След това, следвайки инструкциите, дадени там, трябва да изтеглите изходния код, да изградите и внедрите нашето приложение:

git clone https://github.com/EnMasseProject/enmasse-example-clients
cd enmasse-example-clients/quarkus-example-client
oc new-project myapp
mvn -Pnative -Dfabric8.mode=openshift -Dfabric8.build.strategy=docker package fabric8:build fabric8:resource fabric8:apply

След тези команди приложението ще бъде разгърнато, но няма да стартира, докато не конфигурираме необходимите ни ресурси за съобщения в AMQ Online.

Настройка на системата за съобщения

Сега остава само да зададем ресурсите, от които нашето приложение се нуждае в системата за съобщения. За да направите това, трябва да създадете: 1) адресно пространство за инициализиране на крайната точка на системата за съобщения; 2) адрес за конфигуриране на адресите, които използваме в приложението; 3) Потребител за съобщения за задаване на идентификационни данни на клиента.

Адресно пространство

Обект AddressSpace в AMQ Online е група от адреси, които споделят крайни точки на връзка и правила за удостоверяване и оторизация. Когато създавате адресно пространство, можете да посочите как крайните точки за съобщения ще бъдат изложени:

apiVersion: enmasse.io/v1beta1
kind: AddressSpace
metadata:
  name: quarkus-example
spec:
  type: brokered
  plan: brokered-single-broker
  endpoints:
  - name: messaging
    service: messaging
    exports:
    - name: quarkus-config
      kind: configmap

адрес

Адресите се използват за изпращане и получаване на съобщения. Всеки адрес има тип, който определя неговата семантика, както и план, който определя броя на ресурсите, които трябва да бъдат резервирани. Адресът може да се определи например така:

apiVersion: enmasse.io/v1beta1
kind: Address
metadata:
  name: quarkus-example.orders
spec:
  address: orders
  type: queue
  plan: brokered-queue

Потребител на съобщения

За да сте сигурни, че само надеждни приложения могат да изпращат и получават съобщения до вашите адреси, трябва да създадете потребител в системата за съобщения. За приложения, работещи в клъстер, клиентите могат да бъдат удостоверени с помощта на акаунт за услуга OpenShift. Потребителският "serviceaccount" може да бъде дефиниран например по следния начин:

apiVersion: user.enmasse.io/v1beta1
kind: MessagingUser
metadata:
  name: quarkus-example.app
spec:
  username: system:serviceaccount:myapp:default
  authentication:
    type: serviceaccount
  authorization:
  - operations: ["send", "recv"]
    addresses: ["orders"]

Разрешения за конфигуриране на приложението

За да може AMQ Online да създаде configmap, който използвахме за вграждане на информацията за крайната точка на AMQP, трябва да бъдат зададени Role и RoleBinding:

---
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: quarkus-config
spec:
  rules:
  - apiGroups: [ "" ]
    resources: [ "configmaps" ]
    verbs: [ "create" ]
  - apiGroups: [ "" ]
    resources: [ "configmaps" ]
    resourceNames: [ "quarkus-config" ]
    verbs: [ "get", "update", "patch" ]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: quarkus-config
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: Role
  name: quarkus-config
subjects:
- kind: ServiceAccount
  name: address-space-controller
  namespace: amq-online-infra

Как се прилагат конфигурации

Можете да приложите конфигурацията на системата за съобщения по следния начин:

cd enmasse-example-clients/quarkus-example-client
oc project myapp
oc apply -f src/main/resources/k8s/addressspace
oc apply -f src/main/resources/k8s/address

Проверка на приложението

За да се уверите, че приложението е стартирано, нека първо проверим дали съответните адреси са създадени и активни:

until [[ `oc get address quarkus-example.prices -o jsonpath='{.status.phase}'` == "Active" ]]; do echo "Not yet ready"; sleep 5; done

Тогава нека проверим URL адреса на маршрута на приложението (просто отворете този адрес в браузъра):

echo "http://$(oc get route quarkus-example-client -o jsonpath='{.spec.host}')/prices.html"

Браузърът трябва да показва, че билетите се актуализират периодично, когато съобщенията се изпращат и получават от AMQ Online.

Резюмиране

Така че написахме приложение Quarkus, което използва AMQP за съобщения, конфигурирахме приложението да работи на платформата Red Hat OpenShift и внедрихме конфигурацията му въз основа на AMQ Online конфигурацията. След това създадохме манифестите, необходими за инициализиране на системата за съобщения за нашето приложение.

С това приключваме поредицата за Quarkus, но предстоят много нови и интересни неща, очаквайте!

Източник: www.habr.com