Vert.x Kafka 不遵守 RxJava 线程分配？

Question

给定以下代码：

kafkaConsumer
  .rxSubscription()
  .subscribeOn(Schedulers.io())
  .map(s -> {
    logger.info("Mapping on Thread: " + Thread.currentThread().getName());
    return s;
   })
  .observeOn(Schedulers.computation())
  .subscribe(
     set -> {
       logger.info("Subscribing on Thread: " +Thread.currentThread().getName());
   });

如果kafka消费者是Vert. x Kafka消费者，我希望

.map(s -> {
  logger.info("Mapping on Thread: " + Thread.currentThread().getName());
  return s;
})

会发生在Reactive IO线程上。但是，它在Vert. x事件循环线程上执行。当我运行以下测试类时，相同的场景按照预期在IO线程上运行map方法。

public class ThreadTesting {

public static void main(String args[]) {
  Vertx vertx = Vertx.vertx();
  Observable.fromArray(new String[] {"start"})
    .flatMapSingle(s -> method1())
    .subscribeOn(Schedulers.io())
    .map(
        s -> {
          System.out.println("mapping 2 on Thread: " + Thread.currentThread().getName());
          return s.concat(method2());
        })
    .observeOn(Schedulers.computation())
    .subscribe(
        str -> {
          System.out.println("Subscribing on Thread: " + Thread.currentThread().getName());
        },
        onError -> {
          onError.printStackTrace();
        });
 }

 public static Single<String> method1() {
   System.out.println("Executing method 1 on Thread: " + Thread.currentThread().getName());
   AsyncResultSingle<String> vertxSingle = new AsyncResultSingle<>(
      h -> {
         h.handle(Future.succeededFuture("method 1 string"));
      });
    return vertxSingle;
 }

 public static String method2() {
   System.out.println("Executing method 2 on Thread: " + Thread.currentThread().getName());
   return "method 2 String";
 }
}

是什么导致线程执行中出现这种差异？

Answer 1

Vert.x KafkaConsumer 在事件循环线程上异步发出项目，即使您在 io 调度程序上订阅了它也是如此。

在您的片段中，您尝试强制在计算调度程序上发出项目。它有效，但不适用于您期望的可观察对象：它适用于map操作返回的可观察对象。

如果希望< code>map在< code >计算调度程序上运行，则需要在以下操作之前应用< code>observeOn运算符:

kafkaConsumer
  .rxSubscription()
  .subscribeOn(Schedulers.io())
  .observeOn(Schedulers.computation())
  .map(s -> {
    logger.info("Mapping on Thread: " + Thread.currentThread().getName());
    return s;
   })
  .subscribe(
     set -> {
       logger.info("Subscribing on Thread: " +Thread.currentThread().getName());
   });