Как вызвать нестатическую функцию в remotelisten лямбда-функции ignite-messaging?

Question

У меня есть класс (TopicListenerImp), который прослушивает тему, и я хочу получить сообщение, полученное из функции remoteListen, и передать его в качестве параметра для других нестатических функций в других классах обслуживания.

     @Component
    public class TopicListenerImp implements TopicListener {


      private NotificationService notificationService;
      private SubscriptionRepository subscriptionRepository;
      private SubscriptionRules subscriptionRules;
      private NFInstancesService nfInstancesService;
      private Ignite ignite;

      public TopicListenerImp(
          SubscriptionRules subscriptionRules,
          NotificationService notificationService,
          SubscriptionRepository subscriptionRepository,
          Ignite ignite,
          NFInstancesService nfInstancesService) {
        this.subscriptionRules = subscriptionRules;
        this.notificationService = notificationService;
        this.subscriptionRepository = subscriptionRepository;
        this.nfInstancesService = nfInstancesService;
        this.ignite = ignite;
      }

      @Bean
      public void startTopicListening() {

        IgniteMessaging rmtMsg = ignite.message(ignite.cluster().forLocal());
        rmtMsg.remoteListen(
            "SUSPEND",
            (nodeId, msg) -> {

              notifyIfSubscriptionExist((String) msg); //here where I used the message that comes from topic
              return true; 
            });
      }

      public void notifyIfSubscriptionExist(String msg) {

        List<String> nfInstanceIdSubscriptionId = parseNfInstanceIdSubscriptionId(msg);
        Optional<NFProfile> nfProfile =
            nfInstancesService.getNFInstance(nfInstanceIdSubscriptionId.get(0));
        Optional<SubscriptionData> subscriptionDataOptional =
            subscriptionRepository.getSubscriptionData(nfInstanceIdSubscriptionId.get(1));

        subscriptionDataOptional.ifPresent(
            subscriptionData -> {
              if (subscriptionRules.checkRules(subscriptionData) && nfProfile.isPresent())
                notificationService.sendEventNotification(
                    subscriptionData, nfProfile.get(), NF_PROFILE_CHANGED);
            });
      }

      private List<String> parseNfInstanceIdSubscriptionId(String msg) {
        List<String> values = asList(msg.split(", "));
        return asList(getIdFromMessage(values.get(0)), getIdFromMessage(values.get(1)));
      }

      private String getIdFromMessage(String msg) {
        return msg.substring(msg.indexOf('[') + 1, msg.length() - 1);
      }
    }

Но я получаю следующие ошибки:

    Caused by: org.apache.ignite.binary.BinaryObjectException: Failed to serialize object etc.

Когда я вводю поля статически, это работает. Но когда я делаю это, IDE выдает предупреждение «не обновляйте статическую переменную из метода конструктора».

Answer 1

remoteListen принимает предикат, который будет сериализован и отправлен удаленным узлам по сети. Когда вы вызываете нестатический метод в его реализации, он делает сериализованным весь объект this. Это может привести к неожиданному поведению и множеству данных, отправляемых по сети.

В общем случае не рекомендуется отправлять лямбда-функции по сети, поскольку их сериализация зависит от виртуальной машины и не дает вам контроля над тем, что вына самом деле отправить. Если вы хотите отправить предикат по сети, то создайте класс, реализующий интерфейс предиката, и используйте экземпляр этого класса.

Если вам не нужны все узлы в кластере, чтобы подписаться на тему, тогда localListen будет достаточно.