import com.google.api.core.ApiFutureCallback; //导入依赖的package包/类
@Override
public void flush() {
  List<Span> spans = new ArrayList<>();
  queue.drain(spans::add);
  if (spans.isEmpty()) {
    return;
  }
  PatchTracesRequest request =
      PatchTracesRequest.newBuilder()
          .setProjectId(projectId)
          .setTraces(Traces.newBuilder().addAllTraces(translator.translateSpans(spans)))
          .build();
  ApiFutures.addCallback(
      traceServiceClient.patchTracesCallable().futureCall(request),
      new ApiFutureCallback<Empty>() {
        @Override
        public void onFailure(Throwable t) {
          logger.warn("Error reporting traces.", t);
        }

        @Override
        public void onSuccess(Empty result) {
          logger.info("Successfully reported traces.");
        }
      });
}
 

import com.google.api.core.ApiFutureCallback; //导入依赖的package包/类
private static void doWrite(
    DatabaseReference ref, final boolean shouldSucceed, final boolean shouldTimeout)
    throws InterruptedException {
  final CountDownLatch lock = new CountDownLatch(1);
  final AtomicBoolean success = new AtomicBoolean(false);
  ApiFutures.addCallback(ref.setValueAsync("wrote something"), new ApiFutureCallback<Void>() {
    @Override
    public void onFailure(Throwable throwable) {
      success.compareAndSet(false, false);
      lock.countDown();
    }

    @Override
    public void onSuccess(Void result) {
      success.compareAndSet(false, true);
      lock.countDown();
    }
  });
  boolean finished = lock.await(TestUtils.TEST_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
  if (shouldTimeout) {
    assertTrue("Write finished (expected to timeout).", !finished);
  } else if (shouldSucceed) {
    assertTrue("Write timed out (expected to succeed)", finished);
    assertTrue("Write failed (expected to succeed).", success.get());
  } else {
    assertTrue("Write timed out (expected to fail).", finished);
    assertTrue("Write successful (expected to fail).", !success.get());
  }
}
 

import com.google.api.core.ApiFutureCallback; //导入依赖的package包/类
@Override
public ListenableFuture<String> publish(final String topic, PubsubMessage pubsubMessage) {
	ApiFuture<String> publishFuture =
			this.publisherFactory.createPublisher(topic).publish(pubsubMessage);

	final SettableListenableFuture<String> settableFuture = new SettableListenableFuture<>();
	ApiFutures.addCallback(publishFuture, new ApiFutureCallback<String>() {

		@Override
		public void onFailure(Throwable throwable) {
			LOGGER.warn("Publishing to " + topic + " topic failed.", throwable);
			settableFuture.setException(throwable);
		}

		@Override
		public void onSuccess(String result) {
			if (LOGGER.isDebugEnabled()) {
				LOGGER.debug(
						"Publishing to " + topic + " was successful. Message ID: " + result);
			}
			settableFuture.set(result);
		}

	});

	return settableFuture;
}
 

import com.google.api.core.ApiFutureCallback; //导入依赖的package包/类
@Override
public ListenableFuture<RunResult> doRun() {
  AtomicInteger numPending = new AtomicInteger(batchSize);
  final SettableFuture<RunResult> done = SettableFuture.create();
  String sendTime = String.valueOf(System.currentTimeMillis());
  if (!outstandingBytes.tryAcquire(batchSize * messageSize)) {
    return Futures.immediateFailedFuture(new Exception("Flow control limits reached."));
  }
  for (int i = 0; i < batchSize; i++) {
    ApiFutures.addCallback(publisher
        .publish(
            PubsubMessage.newBuilder()
            .setData(payload)
            .putAttributes("sendTime", sendTime)
            .putAttributes("clientId", id.toString())
            .putAttributes(
                "sequenceNumber", Integer.toString(sequenceNumber.getAndIncrement()))
            .build()), new ApiFutureCallback<String>() {
          @Override
          public void onSuccess(String messageId) {
            outstandingBytes.release(messageSize);
            if (numPending.decrementAndGet() == 0) {
              done.set(RunResult.fromBatchSize(batchSize));
            }
          }

          @Override
          public void onFailure(Throwable t) {
            outstandingBytes.release(messageSize);
            done.setException(t);
          }
        });
  }
  return done;
}
 

import com.google.api.core.ApiFutureCallback; //导入依赖的package包/类
/** Publish messages to a topic.
 * @param args topic name, number of messages
 */
public static void main(String... args) throws Exception {
  // topic id, eg. "my-topic"
  String topicId = args[0];
  int messageCount = Integer.parseInt(args[1]);
  TopicName topicName = TopicName.of(PROJECT_ID, topicId);
  Publisher publisher = null;
  try {
    // Create a publisher instance with default settings bound to the topic
    publisher = Publisher.newBuilder(topicName).build();

    for (int i = 0; i < messageCount; i++) {
      String message = "message-" + i;

      // convert message to bytes
      ByteString data = ByteString.copyFromUtf8(message);
      PubsubMessage pubsubMessage = PubsubMessage.newBuilder()
          .setData(data)
          .build();

      //schedule a message to be published, messages are automatically batched
      ApiFuture<String> future = publisher.publish(pubsubMessage);

      // add an asynchronous callback to handle success / failure
      ApiFutures.addCallback(future, new ApiFutureCallback<String>() {

        @Override
        public void onFailure(Throwable throwable) {
          if (throwable instanceof ApiException) {
            ApiException apiException = ((ApiException) throwable);
            // details on the API exception
            System.out.println(apiException.getStatusCode().getCode());
            System.out.println(apiException.isRetryable());
          }
          System.out.println("Error publishing message : " + message);
        }

        @Override
        public void onSuccess(String messageId) {
          // Once published, returns server-assigned message ids (unique within the topic)
          System.out.println(messageId);
        }
      });
    }
  } finally {
    if (publisher != null) {
      // When finished with the publisher, shutdown to free up resources.
      publisher.shutdown();
    }
  }
}