import com.google.pubsub.v1.SubscriptionName; //导入依赖的package包/类
/**
 * Create a new subscription on Google Cloud Pub/Sub.
 *
 * @param subscriptionName the name of the new subscription
 * @param topicName the name of the topic being subscribed to
 * @param ackDeadline deadline in seconds before a message is resent. If not provided, set to
 *                    default of 10 seconds
 * @param pushEndpoint URL of the service receiving the push messages. If not provided, uses
 *                     message pulling by default
 * @return the created subscription
 */
public Subscription createSubscription(String subscriptionName, String topicName,
		Integer ackDeadline, String pushEndpoint) {
	Assert.hasText(subscriptionName, "No subscription name was specified.");
	Assert.hasText(topicName, "No topic name was specified.");

	int finalAckDeadline = this.defaultAckDeadline;
	if (ackDeadline != null) {
		Assert.isTrue(ackDeadline >= 0,
				"The acknowledgement deadline value can't be negative.");
		finalAckDeadline = ackDeadline;
	}

	PushConfig.Builder pushConfigBuilder = PushConfig.newBuilder();
	if (pushEndpoint != null) {
		pushConfigBuilder.setPushEndpoint(pushEndpoint);
	}

	return this.subscriptionAdminClient.createSubscription(
			SubscriptionName.of(this.projectId, subscriptionName),
			TopicName.of(this.projectId, topicName),
			pushConfigBuilder.build(),
			finalAckDeadline);
}
 

import com.google.pubsub.v1.SubscriptionName; //导入依赖的package包/类
@Override
public PullRequest createPullRequest(String subscriptionName, Integer maxMessages,
		Boolean returnImmediately) {
	Assert.hasLength(subscriptionName, "The subscription name must be provided.");

	PullRequest.Builder pullRequestBuilder =
			PullRequest.newBuilder().setSubscriptionWithSubscriptionName(
					SubscriptionName.of(this.projectId, subscriptionName));

	if (maxMessages != null) {
		pullRequestBuilder.setMaxMessages(maxMessages);
	}

	if (returnImmediately != null) {
		pullRequestBuilder.setReturnImmediately(returnImmediately);
	}

	return pullRequestBuilder.build();
}
 

import com.google.pubsub.v1.SubscriptionName; //导入依赖的package包/类
/**
 * Get the configuration of a Google Cloud Pub/Sub subscription.
 *
 * @param subscriptionName canonical subscription name, e.g., "subscriptionName"
 * @return subscription configuration or {@code null} if subscription doesn't exist
 */
public Subscription getSubscription(String subscriptionName) {
	Assert.hasText(subscriptionName, "No subscription name was specified");

	try {
		return this.subscriptionAdminClient.getSubscription(
				SubscriptionName.create(this.projectId, subscriptionName));
	}
	catch (ApiException aex) {
		if (aex.getStatusCode().getCode() == StatusCode.Code.NOT_FOUND) {
			return null;
		}

		throw aex;
	}
}
 

import com.google.pubsub.v1.SubscriptionName; //导入依赖的package包/类
/** Receive messages over a subscription. */
public static void main(String... args) throws Exception {
  // set subscriber id, eg. my-sub
  String subscriptionId = args[0];
  SubscriptionName subscriptionName = SubscriptionName.of(PROJECT_ID, subscriptionId);
  Subscriber subscriber = null;
  try {
    // create a subscriber bound to the asynchronous message receiver
    subscriber =
        Subscriber.newBuilder(subscriptionName, new MessageReceiverExample()).build();
    subscriber.startAsync().awaitRunning();
    // Continue to listen to messages
    while (true) {
      PubsubMessage message = messages.take();
      System.out.println("Message Id: " + message.getMessageId());
      System.out.println("Data: " + message.getData().toStringUtf8());
    }
  } finally {
    if (subscriber != null) {
      subscriber.stopAsync();
    }
  }
}
 

import com.google.pubsub.v1.SubscriptionName; //导入依赖的package包/类
@Override
public Subscriber createSubscriber(String subscriptionName, MessageReceiver receiver) {
	Subscriber.Builder subscriberBuilder = Subscriber.newBuilder(
			SubscriptionName.of(this.projectId, subscriptionName), receiver);

	if (this.channelProvider != null) {
		subscriberBuilder.setChannelProvider(this.channelProvider);
	}

	if (this.executorProvider != null) {
		subscriberBuilder.setExecutorProvider(this.executorProvider);
	}

	if (this.credentialsProvider != null) {
		subscriberBuilder.setCredentialsProvider(this.credentialsProvider);
	}

	if (this.headerProvider != null) {
		subscriberBuilder.setHeaderProvider(this.headerProvider);
	}

	if (this.systemExecutorProvider != null) {
		subscriberBuilder.setSystemExecutorProvider(this.systemExecutorProvider);
	}

	if (this.flowControlSettings != null) {
		subscriberBuilder.setFlowControlSettings(this.flowControlSettings);
	}

	if (this.maxAckDurationPeriod != null) {
		subscriberBuilder.setMaxAckExtensionPeriod(this.maxAckDurationPeriod);
	}

	if (this.parallelPullCount != null) {
		subscriberBuilder.setParallelPullCount(this.parallelPullCount);
	}

	return subscriberBuilder.build();
}
 

import com.google.pubsub.v1.SubscriptionName; //导入依赖的package包/类
/**
 * Delete a subscription from Google Cloud Pub/Sub.
 *
 * @param subscriptionName
 */
public void deleteSubscription(String subscriptionName) {
	Assert.hasText(subscriptionName, "No subscription name was specified");

	this.subscriptionAdminClient.deleteSubscription(
			SubscriptionName.of(this.projectId, subscriptionName));
}
 

import com.google.pubsub.v1.SubscriptionName; //导入依赖的package包/类
public void forceDeleteSubscription(QueueSubscriber qsubscriber) {
    SubscriptionName subscriptionName = SubscriptionName.create(projectId, qsubscriber.getSubscriberId());
    try (SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create()) {
        subscriptionAdminClient.deleteSubscription(subscriptionName);
    } catch (Exception ioe) {
        System.err.println(ExceptionToString.format(ioe));
    }
}
 

import com.google.pubsub.v1.SubscriptionName; //导入依赖的package包/类
private CPSSubscriberTask(StartRequest request) {
  super(request, "gcloud", MetricsHandler.MetricName.END_TO_END_LATENCY);
  this.subscription =
      SubscriptionName.create(request.getProject(), request.getPubsubOptions().getSubscription());
  try {
    this.subscriber =
        Subscriber.defaultBuilder(this.subscription, this)
            .setParallelPullCount(Runtime.getRuntime().availableProcessors() * 5)
            .build();
  } catch (Exception e) {
    throw new RuntimeException(e);
  }
}
 

import com.google.pubsub.v1.SubscriptionName; //导入依赖的package包/类
/**
 * Create a pull subscription.
 *
 * @param args topic subscriptionId
 * @throws Exception exception thrown if operation is unsuccessful
 */
public static void main(String... args) throws Exception {

  // Your Google Cloud Platform project ID
  String projectId = ServiceOptions.getDefaultProjectId();

  // Your topic ID, eg. "my-topic"
  String topicId = args[0];

  // Your subscription ID eg. "my-sub"
  String subscriptionId = args[1];

  TopicName topicName = TopicName.of(projectId, topicId);

  // Create a new subscription
  SubscriptionName subscriptionName = SubscriptionName.of(projectId, subscriptionId);
  try (SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create()) {
    // create a pull subscription with default acknowledgement deadline (= 10 seconds)
    Subscription subscription =
        subscriptionAdminClient.createSubscription(
            subscriptionName, topicName, PushConfig.getDefaultInstance(), 0);
  } catch (ApiException e) {
    // example : code = ALREADY_EXISTS(409) implies subscription already exists
    System.out.print(e.getStatusCode().getCode());
    System.out.print(e.isRetryable());
  }

  System.out.printf(
      "Subscription %s:%s created.\n",
      subscriptionName.getProject(), subscriptionName.getSubscription());
}
 

import com.google.pubsub.v1.SubscriptionName; //导入依赖的package包/类
private void deleteTestSubscription() throws Exception {
  try (SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create()) {
    subscriptionAdminClient.deleteSubscription(
        SubscriptionName.of(projectId, subscriptionId));
  } catch (IOException e) {
    System.err.println("Error deleting subscription " + e.getMessage());
  }
}
 

import com.google.pubsub.v1.SubscriptionName; //导入依赖的package包/类
@Override
public void produce(Map<String, String> lastOffsets, int maxBatchSize) throws StageException {
  SynchronousQueue<MessageReplyConsumerBundle> workQueue = new SynchronousQueue<>();

  SubscriptionName subscriptionName = SubscriptionName.create(conf.credentials.projectId, conf.subscriptionId);

  executor = Executors.newFixedThreadPool(getNumberOfThreads());

  for (int i = 0; i < conf.maxThreads; i++) {
    MessageProcessor messageProcessor = new MessageProcessorImpl(
        getContext(),
        Math.min(maxBatchSize, conf.basic.maxBatchSize),
        conf.basic.maxWaitTime,
        parserFactory,
        workQueue
    );
    executor.submit(messageProcessor);
    messageProcessors.add(messageProcessor);
  }

  ExecutorProvider executorProvider = InstantiatingExecutorProvider.newBuilder()
      .setExecutorThreadCount(conf.advanced.numThreadsPerSubscriber)
      .build();

  ChannelProvider channelProvider = getChannelProvider();
  FlowControlSettings flowControlSettings = getFlowControlSettings();

  for (int i = 0; i < conf.advanced.numSubscribers; i++) {
    Subscriber s = Subscriber.defaultBuilder(subscriptionName, new MessageReceiverImpl(workQueue))
        .setCredentialsProvider(credentialsProvider)
        .setExecutorProvider(executorProvider)
        .setChannelProvider(channelProvider)
        .setFlowControlSettings(flowControlSettings)
        .build();
    s.addListener(new Subscriber.Listener() {
      @Override
      public void failed(Subscriber.State from, Throwable failure) {
        LOG.error("Exception thrown in Subscriber: {}", failure.toString(), failure);
        LOG.error("Subscriber state: {}", from.toString());
        Throwables.propagate(failure);
      }
    }, MoreExecutors.directExecutor());
    subscribers.add(s);
  }

  try {
    subscribers.forEach(Subscriber::startAsync);
  } finally {
    LOG.info("Started {} subscribers.", conf.maxThreads);
  }

  while (!getContext().isStopped()) {
    ThreadUtil.sleep(1000);
  }
}