google-cloud-pubsub - 批处理 PubSub 请求

Question

批处理 pubsub 请求的 NODEJS 示例代码如下所示:

// Imports the Google Cloud client library
const PubSub = require(`@google-cloud/pubsub`);

// Creates a client
const pubsub = new PubSub();

/**
 * TODO(developer): Uncomment the following lines to run the sample.
 */
// const topicName = 'your-topic';
// const data = JSON.stringify({ foo: 'bar' });
// const maxMessages = 10;
// const maxWaitTime = 10000;

// Publishes the message as a string, e.g. "Hello, world!" or JSON.stringify(someObject)
const dataBuffer = Buffer.from(data);

pubsub
  .topic(topicName)
  .publisher({
    batching: {
      maxMessages: maxMessages,
      maxMilliseconds: maxWaitTime,
    },
  })
  .publish(dataBuffer)
  .then(results => {
    const messageId = results[0];
    console.log(`Message ${messageId} published.`);
  })
  .catch(err => {
    console.error('ERROR:', err);
  });

对我来说，不清楚如何使用这个例子同时发布多条消息。有人可以解释如何调整此代码以便它可以用于同时发布多条消息吗？

Answer 1

如果你想批处理消息，那么你需要保持发布者并多次调用 publish 。例如，您可以将代码更改为如下所示:

// Imports the Google Cloud client library
const PubSub = require(`@google-cloud/pubsub`);

// Creates a client
const pubsub = new PubSub();


const topicName = 'my-topic';
const maxMessages = 10;
const maxWaitTime = 10000;
const data1 = JSON.stringify({ foo: 'bar1' });
const data2 = JSON.stringify({ foo: 'bar2' });
const data3 = JSON.stringify({ foo: 'bar3' });

const publisher = pubsub.topic(topicName).publisher({
    batching: {
      maxMessages: maxMessages,
      maxMilliseconds: maxWaitTime,
    },
  })

function handleResult(p) {
  p.then(results => {
    console.log(`Message ${results} published.`);
  })
  .catch(err => {
    console.error('ERROR:', err);
  });
}

// Publish three messages
handleResult(publisher.publish(Buffer.from(data1)));
handleResult(publisher.publish(Buffer.from(data2)));
handleResult(publisher.publish(Buffer.from(data3)));

消息的批处理由 maxMessages 和 maxMilliseconds 属性处理。前者表示批处理中包含的最大消息数。后者表示等待发布批次的最大毫秒数。这些属性在更大的批次(可能更有效)与发布延迟之间进行权衡。如果您要快速发布很多消息，那么 maxMilliseconds 属性不会有太大影响；一旦十条消息准备就绪，客户端库就会向 Cloud Pub/Sub 服务发出发布请求。但是，如果发布是零星的或缓慢的，则可能会在发送十条消息之前发送一批消息。

在上面的示例代码中，我们对三个消息调用了 publish。这不足以填满一个批次并发送它。因此，在第一次调用 publish 10,000 毫秒后，这三个消息将作为批处理发送到 Cloud Pub/Sub。