问急切地停止Akka流源

EN

static Subscriber<Object> println() {
    return new Subscriber<Object>() {

        Subscription s;

        int n;

        @Override
        public void onSubscribe(Subscription s) {
            this.s = s;
            s.request(5);
        }

        @Override
        public void onNext(Object t) {
            System.out.println(Thread.currentThread().getName() 
                  + ": " + t + " - " + (++n));
            if (s != null) {
                s.cancel();
                s = null;
            }
        }

        @Override
        public void onError(Throwable t) {
            t.printStackTrace();
        }

        @Override
        public void onComplete() {
            System.out.println(Thread.currentThread().getName() + ": DONE");
        }
    };
}

public static void main(String[] args) throws Exception {
    Config cfg = ConfigFactory.parseResources(
        AkkaRange.class, "/akka-streams.conf").resolve();
    ActorSystem actorSystem = ActorSystem.create("sys", cfg);

    ActorMaterializer materializer = ActorMaterializer.create(actorSystem);

    Source<Integer, NotUsed> source = Source.repeat(1);

    Publisher<Integer> p = source.runWith(Sink.asPublisher(
         AsPublisher.WITH_FANOUT), materializer);

    p.subscribe(println());

    Thread.sleep(1000);

    actorSystem.terminate();
}