code: public void launch() throws Exception {
          if (init.get())
              return;
          // Register a SIGINT handler for graceful shutdown.
          if (!init.get())
              init();
  
          log.info("Subscribing to " + channel + " on stream Id " + streamId);
          log.info("Using aeron directory " + ctx.aeronDirectoryName());
  
          // Register a SIGINT handler for graceful shutdown.
          SigInt.register(() -> running.set(false));
  
          // Create an Aeron instance with client-provided context configuration, connect to the
          // media driver, and add a subscription for the given channel and stream using the supplied
          // dataHandler method, which will be called with new messages as they are received.
          // The Aeron and Subscription classes implement AutoCloseable, and will automatically
          // clean up resources when this try block is finished.
          //Note here that we are either creating 1 or 2 subscriptions.
          //The first one is a  normal 1 subscription listener.
          //The second one is when we want to send responses
  
          if (channel == null)
              throw new IllegalStateException("No channel for subscriber defined");
          if (streamId <= 0)
              throw new IllegalStateException("No stream for subscriber defined");
          if (aeron == null)
              throw new IllegalStateException("No aeron instance defined");
          boolean started = false;
          while (!started) {
              try (final Subscription subscription = aeron.addSubscription(channel, streamId)) {
                  this.subscription = subscription;
                  log.info("Beginning subscribe on channel " + channel + " and stream " + streamId);
                  AeronUtil.subscriberLoop(new FragmentAssembler(new NDArrayFragmentHandler(ndArrayCallback)),
                                  fragmentLimitCount, running, launched).accept(subscription);
                  started = true;
  
              } catch (Exception e) {
                  log.warn("Unable to connect...trying again on channel " + channel, e);
              }
          }
  
      }
docs: Launch a background thread
  that subscribes to  the aeron context
  @throws Exception