Облачно-ориентиран обмен на съобщения на платформата Red Hat OpenShift с използване на Quarkus и AMQ Online

Здравейте на всички! Ето го – нашият заключителен пост от серията за Quarkus! (Между другото, вижте уебинара ни «Това е Quarkus – Kubernetes native Java фреймворк». Ще покажем как да започнете «от нула» или да пренесете готови решения)

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

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

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

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

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

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

Приложение Quarkus

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

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

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

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

Обработчик на поръчки

Обработчикът на поръчки е още по-прост, той просто връща идентификатор на потвърждение на адрес "confirmations".

@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 orders;
 
    @GET
    @Produces(MediaType.TEXT_PLAIN)
    public String hello() {
        return "hello";
    }
 
 
    @GET
    @Path("/stream")
    @Produces(MediaType.SERVER_SENT_EVENTS)
    public Publisher stream() {
        return orders;
    }
}

Настройка

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

Конектори

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

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

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

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

По време на компилацията името на хоста и номера на порта за AMQP крайната точка не са известни, затова те трябва да бъдат инжектирани. Крайната точка може да бъде зададена в configmap, създаден от 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 propertyNames;
 
    static {
        propertyNames = new HashSet();
        propertyNames.add("amqp-username");
        propertyNames.add("amqp-password");
    }
 
    @Override
    public Set getPropertyNames() {
        return propertyNames;
    }
 
    @Override
    public Map getProperties() {
        try {
            Map 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 "Все още не е готово"; 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, но ни очакват много новини и интересни неща, останете с нас!

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