Google pubsub : 교수형 자바 샘플 클라이언트 걸기

Question

google pubsub를 사용하여 가입 된 주제의 메시지를 비동기식으로 가져옵니다. 나는 아래의 샘플 코드를 사용하고

:

public void startStreaming() { 
     Subscriber subscriber = null; 
     try { 
      SubscriptionName subscription = SubscriptionName.of("topic-test", 
        "subscription-id-1234"); 
      ExecutorProvider executorProvider = 
        InstantiatingExecutorProvider.newBuilder() 
          .setExecutorThreadCount(1) 
          .build(); 

      MessageReceiver receiver = 
        new MessageReceiver() { 
         @Override 
         public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) { 
          try { 
           String msg = message.getData().toStringUtf8(); 
           System.out.println(msg); 
          } catch (Exception e) { 
           throw new RuntimeException("Failed to pull messages from topic"); 
          } 
          consumer.ack(); 
         } 
        }; 
      subscriber = Subscriber 
        .newBuilder(subscription, receiver) 
        .setExecutorProvider(executorProvider) 
        .build(); 
      subscriber.addListener(
        new Subscriber.Listener() { 
         @Override 
         public void failed(Subscriber.State from, Throwable failure) { 
          // Handle failure. This is called when the Subscriber encountered a fatal error and is shutting down. 
          logger.error("Subscriber is not able to hook on google pubsub topic", failure); 
         } 
        }, 
        MoreExecutors.directExecutor()); 
      subscriber.startAsync().awaitRunning(); 
      // why should I put sleep here?? 
      Thread.sleep(60000); 
     } catch (Exception e) { 
     } finally { 
      if (subscriber != null) { 
       subscriber.stopAsync().awaitTerminated(); 
      } 
     } 
    } 

그것은 몇 분 동안 (메시지를 섭취) 잘 작동 및 메시지에 더 이상 섭취에 박히면서, 나는 내가 다시 시작 한 번으로 메시지가 있다는 것을 확신 당기는 클라이언트는 데이터를 정상적으로 읽습니다.

Nov 12, 2017 7:33:17 PM com.google.cloud.pubsub.v1.StreamingSubscriberConnection sendAckOperations 
WARNING: failed to send acks 
java.lang.IllegalStateException: call was cancelled 
    at com.google.common.base.Preconditions.checkState(Preconditions.java:444) 
    at io.grpc.internal.ClientCallImpl.sendMessage(ClientCallImpl.java:405) 
    at io.grpc.ForwardingClientCall.sendMessage(ForwardingClientCall.java:52) 
    at io.grpc.ForwardingClientCall.sendMessage(ForwardingClientCall.java:52) 
    at io.grpc.stub.ClientCalls$CallToStreamObserverAdapter.onNext(ClientCalls.java:286) 
    at com.google.cloud.pubsub.v1.StreamingSubscriberConnection.sendAckOperations(StreamingSubscriberConnection.java:274) 
    at com.google.cloud.pubsub.v1.MessageDispatcher.processOutstandingAckOperations(MessageDispatcher.java:602) 
    at com.google.cloud.pubsub.v1.MessageDispatcher.access$2100(MessageDispatcher.java:56) 
    at com.google.cloud.pubsub.v1.MessageDispatcher$AckDeadlineAlarm.run(MessageDispatcher.java:529) 
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) 
    at java.util.concurrent.FutureTask.run(FutureTask.java:266) 
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) 
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) 
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 
    at java.lang.Thread.run(Thread.java:745) 

그리고 ExecutorProvider executorProvider를 사용하지 않을 때 나는 아래의 로그

2017-11-12 19:08:43 [grpc-default-worker-ELG-2-3] DEBUG io.grpc.netty.NettyClientHandler [id: 0xdc0e7c4d, L:/10.8.17.4:47347 - R:pubsub.googleapis.com/1.1.1.1:111] OUTBOUND WINDOW_UPDATE: streamId=3 windowSizeIncrement=1048576 
2017-11-12 19:08:43 [grpc-default-worker-ELG-2-4] DEBUG io.grpc.netty.NettyClientHandler [id: 0x330498ce, L:/10.8.17.4:47346 - R:pubsub.googleapis.com/1.1.1.1:111] OUTBOUND GO_AWAY: lastStreamId=0 errorCode=0 length=0 bytes= 
2017-11-12 19:08:43 [grpc-default-worker-ELG-2-3] DEBUG io.grpc.netty.NettyClientHandler [id: 0xdc0e7c4d, L:/10.8.17.4:47347 - R:pubsub.googleapis.com/1.1.1.1:111] OUTBOUND GO_AWAY: lastStreamId=0 errorCode=0 length=0 bytes= 
2017-11-12 19:08:44 [threadDeathWatcher-1-1] DEBUG io.netty.buffer.PoolThreadCache Freed 3 thread-local buffer(s) from thread: threadDeathWatcher-1-1 
2017-11-12 19:08:44 [threadDeathWatcher-1-1] DEBUG io.netty.buffer.PoolThreadCache Freed 1 thread-local buffer(s) from thread: threadDeathWatcher-1-1 
2017-11-12 19:08:44 [grpc-default-worker-ELG-2-3] DEBUG io.netty.buffer.PoolThreadCache Freed 153 thread-local buffer(s) from thread: grpc-default-worker-ELG-2-3 
2017-11-12 19:08:44 [grpc-default-worker-ELG-2-4] DEBUG io.netty.buffer.PoolThreadCache Freed 131 thread-local buffer(s) from thread: grpc-default-worker-ELG-2-4 

당신이 도와주세요 수있어? 그것은 지금 잘 작동

Answer 1

이

lines, 

//   if (subscriber != null) { 
//    subscriber.stopAsync().awaitTerminated(); 
//   } 

설명 아래를 제거했다, 여기 github에 thread 1827

입니다 그리고 그것은 정말 이해가 무엇 누락 된 구글 클라우드에 더 많은 문서입니다 docs는 github 스레드에서 제공된 주석과 같은 자세한 내용을 제공합니다.

편집 : Github의 티켓이 더 문서를 추가하기 위해 만든, 아래 닫혀 : 2616