Fan-out delivers one published message to many independent consumers. A single Publish writes the message once to the topic's Events Store log gcp.{topic}, then the connector fans out one copy to each subscription's Queue channel gcp.sub.{subscription} — applying that subscription's filter. Every subscription has its own backlog and its own ack state, so a slow or filtered consumer never blocks the others. This is the classic "one event, many consumers" pattern: an order-placed event reaches a billing subscription, a shipping subscription, and an analytics subscription from a single publish.

Overview

CreateTopic once, then CreateSubscription for each consumer that should receive a copy. Publish resolves, at publish time, to every subscription bound to the topic and enqueues one copy per subscription (skipping ones whose filter does not match, and detached ones). Each subscription is pulled and acknowledged independently.

Step	Action	Behavior
Topic	`CreateTopic("events")`	Events Store log `gcp.events`
Subscriptions	`CreateSubscription` ×N	Each binds a Queue channel `gcp.sub.{name}`
Publish	`Publish`	One write to `gcp.events`; one Queue copy fanned out per subscription
Filtered subscription	`CreateSubscription(filter=…)`	Receives the copy only when its filter matches
Per-subscription pull	`Pull` / `StreamingPull`	Independent backlog + ack state per subscription

Fan-out semantics:

One server-assigned message id per publish, the same across every subscription's copy.
Zero subscriptions → the publish still succeeds and the message lands only in the topic log (no error).
Filters apply at fan-out time — a non-matching subscription is simply never enqueued (≈ auto-acked), and a publish that matches no subscription still succeeds.

How it works

A single publish is written once to the topic log gcp.{topic}, then the connector fans out one Queue copy to every subscription bound to the topic. Each subscription has its own queue, filter, and ack state; a filtered subscription receives the copy only when its filter matches.

One publish writes once to the topic log gcp.{topic} and fans out an independent Queue copy to every subscription on gcp.sub.{subscription}, all sharing one message id; a filtered subscription receives its copy only when the message attributes match its filter.

Fan one publish to many subscriptions

Create a topic, attach two subscriptions, publish once, and watch both subscriptions receive the same message id from their own independent queues. Each client sets only PUBSUB_EMULATOR_HOST (default localhost:8085) and a project id.

package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"time"

	"cloud.google.com/go/pubsub"
	"cloud.google.com/go/pubsub/apiv1/pubsubpb"
)

func projectID() string {
	if v := os.Getenv("PUBSUB_PROJECT_ID"); v != "" {
		return v
	}
	return "my-project"
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()

	// PUBSUB_EMULATOR_HOST routes the official client at the connector (insecure gRPC).
	client, err := pubsub.NewClient(ctx, projectID())
	if err != nil {
		log.Fatalf("NewClient: %v", err)
	}
	defer client.Close()

	// One topic, two independent subscriptions → gcp.sub.billing / gcp.sub.shipping.
	topic, err := client.CreateTopic(ctx, "events")
	if err != nil {
		log.Fatalf("CreateTopic: %v", err)
	}
	defer topic.Stop()
	for _, name := range []string{"billing", "shipping"} {
		if _, err := client.CreateSubscription(ctx, name, pubsub.SubscriptionConfig{
			Topic:       topic,
			AckDeadline: 10 * time.Second,
		}); err != nil {
			log.Fatalf("CreateSubscription %q: %v", name, err)
		}
	}

	// One Publish fans out to both subscriptions with a shared message id.
	msgID, err := topic.Publish(ctx, &pubsub.Message{Data: []byte("order #1001 placed")}).Get(ctx)
	if err != nil {
		log.Fatalf("Publish: %v", err)
	}
	fmt.Printf("published: %s\n", msgID)

	// Each subscription has its own backlog — pull from both.
	subClient, err := pubsub.NewSubscriberClient(ctx)
	if err != nil {
		log.Fatalf("NewSubscriberClient: %v", err)
	}
	defer subClient.Close()
	for _, name := range []string{"billing", "shipping"} {
		subPath := fmt.Sprintf("projects/%s/subscriptions/%s", projectID(), name)
		resp, err := subClient.Pull(ctx, &pubsubpb.PullRequest{Subscription: subPath, MaxMessages: 1})
		if err != nil {
			log.Fatalf("Pull %q: %v", name, err)
		}
		rm := resp.GetReceivedMessages()[0]
		fmt.Printf("%s received message id=%s\n", name, rm.GetMessage().GetMessageId())
		_ = subClient.Acknowledge(ctx, &pubsubpb.AcknowledgeRequest{
			Subscription: subPath, AckIds: []string{rm.GetAckId()},
		})
	}
}

import os

from google.cloud import pubsub_v1


def project_id() -> str:
    return os.environ.get("PUBSUB_PROJECT_ID", "my-project")


def main() -> None:
    publisher = pubsub_v1.PublisherClient()
    subscriber = pubsub_v1.SubscriberClient()
    proj = project_id()
    topic_path = publisher.topic_path(proj, "events")  # -> gcp.events

    # One topic, two independent subscriptions → gcp.sub.billing / gcp.sub.shipping.
    publisher.create_topic(request={"name": topic_path})
    sub_paths = {}
    for name in ("billing", "shipping"):
        sub_path = subscriber.subscription_path(proj, name)
        subscriber.create_subscription(request={"name": sub_path, "topic": topic_path})
        sub_paths[name] = sub_path

    # One Publish fans out to both subscriptions with a shared message id.
    msg_id = publisher.publish(topic_path, b"order #1001 placed").result(timeout=15)
    print(f"published: {msg_id}")

    # Each subscription has its own backlog.
    for name, sub_path in sub_paths.items():
        resp = subscriber.pull(request={"subscription": sub_path, "max_messages": 1}, timeout=20)
        rm = resp.received_messages[0]
        assert rm.message.message_id == msg_id  # same message id across subscriptions
        print(f"{name} received message id={rm.message.message_id}")
        subscriber.acknowledge(request={"subscription": sub_path, "ack_ids": [rm.ack_id]})
    subscriber.close()


if __name__ == "__main__":
    main()

import com.google.api.gax.core.NoCredentialsProvider;
import com.google.api.gax.grpc.GrpcTransportChannel;
import com.google.api.gax.rpc.FixedTransportChannelProvider;
import com.google.api.gax.rpc.TransportChannelProvider;
import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
import com.google.cloud.pubsub.v1.SubscriptionAdminSettings;
import com.google.cloud.pubsub.v1.TopicAdminClient;
import com.google.cloud.pubsub.v1.TopicAdminSettings;
import com.google.cloud.pubsub.v1.stub.GrpcSubscriberStub;
import com.google.cloud.pubsub.v1.stub.SubscriberStubSettings;
import com.google.protobuf.ByteString;
import com.google.pubsub.v1.AcknowledgeRequest;
import com.google.pubsub.v1.PublishRequest;
import com.google.pubsub.v1.PubsubMessage;
import com.google.pubsub.v1.PullRequest;
import com.google.pubsub.v1.PullResponse;
import com.google.pubsub.v1.PushConfig;
import com.google.pubsub.v1.ReceivedMessage;
import com.google.pubsub.v1.SubscriptionName;
import com.google.pubsub.v1.TopicName;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import java.util.List;

public final class Main {
    public static void main(String[] args) throws Exception {
        String emulatorHost = System.getenv().getOrDefault("PUBSUB_EMULATOR_HOST", "localhost:8085");
        String projectId = System.getenv().getOrDefault("PUBSUB_PROJECT_ID", "my-project");

        ManagedChannel channel = ManagedChannelBuilder.forTarget(emulatorHost).usePlaintext().build();
        TransportChannelProvider channelProvider =
                FixedTransportChannelProvider.create(GrpcTransportChannel.create(channel));
        NoCredentialsProvider noCreds = NoCredentialsProvider.create();

        TopicName topic = TopicName.of(projectId, "events"); // -> gcp.events

        try (TopicAdminClient topicAdmin = TopicAdminClient.create(TopicAdminSettings.newBuilder()
                .setTransportChannelProvider(channelProvider).setCredentialsProvider(noCreds).build());
                SubscriptionAdminClient subAdmin = SubscriptionAdminClient.create(
                        SubscriptionAdminSettings.newBuilder()
                                .setTransportChannelProvider(channelProvider)
                                .setCredentialsProvider(noCreds).build());
                GrpcSubscriberStub subStub = GrpcSubscriberStub.create(SubscriberStubSettings.newBuilder()
                        .setTransportChannelProvider(channelProvider)
                        .setCredentialsProvider(noCreds).build())) {

            // One topic, two independent subscriptions.
            topicAdmin.createTopic(topic);
            for (String name : List.of("billing", "shipping")) {
                subAdmin.createSubscription(SubscriptionName.of(projectId, name), topic,
                        PushConfig.getDefaultInstance(), 10);
            }

            // One Publish fans out to both subscriptions with a shared message id.
            String msgId = topicAdmin.publish(PublishRequest.newBuilder().setTopic(topic.toString())
                    .addMessages(PubsubMessage.newBuilder()
                            .setData(ByteString.copyFromUtf8("order #1001 placed")).build())
                    .build()).getMessageIds(0);
            System.out.printf("published: %s%n", msgId);

            // Each subscription has its own backlog.
            for (String name : List.of("billing", "shipping")) {
                String subPath = SubscriptionName.of(projectId, name).toString();
                PullResponse resp = subStub.pullCallable().call(PullRequest.newBuilder()
                        .setSubscription(subPath).setMaxMessages(1).build());
                ReceivedMessage got = resp.getReceivedMessages(0);
                System.out.printf("%s received message id=%s%n", name, got.getMessage().getMessageId());
                subStub.acknowledgeCallable().call(AcknowledgeRequest.newBuilder()
                        .setSubscription(subPath).addAckIds(got.getAckId()).build());
            }
        } finally {
            channel.shutdown();
        }
    }
}

import { PubSub, v1 } from "@google-cloud/pubsub";

const projectId = process.env["PUBSUB_PROJECT_ID"] ?? "my-project";
const baseOptions = new PubSub({ projectId }).options;
const options = { ...baseOptions, port: baseOptions.port === undefined ? undefined : Number(baseOptions.port) };
const publisher = new v1.PublisherClient(options);
const subscriber = new v1.SubscriberClient(options);

async function main(): Promise<void> {
  const topic = publisher.projectTopicsPath(projectId, "events"); // -> gcp.events

  // One topic, two independent subscriptions → gcp.sub.billing / gcp.sub.shipping.
  await publisher.createTopic({ name: topic });
  const subNames = ["billing", "shipping"].map((n) => subscriber.subscriptionPath(projectId, n));
  for (const sub of subNames) {
    await subscriber.createSubscription({ name: sub, topic, ackDeadlineSeconds: 10 });
  }

  // One Publish fans out to both subscriptions with a shared message id.
  const [published] = await publisher.publish({
    topic,
    messages: [{ data: Buffer.from("order #1001 placed") }],
  });
  const msgId = published.messageIds?.[0] ?? "";
  console.log(`published: ${msgId}`);

  // Each subscription has its own backlog.
  for (const sub of subNames) {
    const [pull] = await subscriber.pull({ subscription: sub, maxMessages: 1 });
    const rm = pull.receivedMessages![0];
    console.log(`${sub.split("/").pop()} received message id=${rm.message!.messageId}`);
    await subscriber.acknowledge({ subscription: sub, ackIds: [rm.ackId!] });
  }
}

main().catch((err) => {
  console.error(err);
  process.exit(1);
});

using Google.Api.Gax;
using Google.Cloud.PubSub.V1;
using Google.Protobuf;

var projectId = Environment.GetEnvironmentVariable("PUBSUB_PROJECT_ID") ?? "my-project";
var topicName = TopicName.FromProjectTopic(projectId, "events"); // -> gcp.events

// The .NET client does NOT auto-detect the emulator — set EmulatorOnly.
var publisher = await new PublisherServiceApiClientBuilder
{
    EmulatorDetection = EmulatorDetection.EmulatorOnly,
}.BuildAsync();
var subscriber = await new SubscriberServiceApiClientBuilder
{
    EmulatorDetection = EmulatorDetection.EmulatorOnly,
}.BuildAsync();

// One topic, two independent subscriptions → gcp.sub.billing / gcp.sub.shipping.
await publisher.CreateTopicAsync(topicName);
var subNames = new[] { "billing", "shipping" }
    .Select(n => SubscriptionName.FromProjectSubscription(projectId, n)).ToArray();
foreach (var sub in subNames)
{
    await subscriber.CreateSubscriptionAsync(sub, topicName, pushConfig: null, ackDeadlineSeconds: 10);
}

// One Publish fans out to both subscriptions with a shared message id.
var publishResponse = await publisher.PublishAsync(topicName, new[]
{
    new PubsubMessage { Data = ByteString.CopyFromUtf8("order #1001 placed") },
});
var msgId = publishResponse.MessageIds[0];
Console.WriteLine($"published: {msgId}");

// Each subscription has its own backlog.
foreach (var sub in subNames)
{
    var pull = await subscriber.PullAsync(sub, maxMessages: 1);
    var rm = pull.ReceivedMessages[0];
    Console.WriteLine($"{sub.SubscriptionId} received message id={rm.Message.MessageId}");
    await subscriber.AcknowledgeAsync(sub, new[] { rm.AckId });
}

# frozen_string_literal: true

require "google/cloud/pubsub"

project_id = ENV["PUBSUB_PROJECT_ID"] || "my-project"
emulator_host = ENV["PUBSUB_EMULATOR_HOST"] || "localhost:8085"

pubsub = Google::Cloud::PubSub.new(project_id: project_id, emulator_host: emulator_host)
topic_admin = pubsub.topic_admin
sub_admin   = pubsub.subscription_admin
topic_path  = pubsub.topic_path("events") # -> gcp.events

# One topic, two independent subscriptions → gcp.sub.billing / gcp.sub.shipping.
topic = topic_admin.create_topic(name: topic_path)
sub_paths = %w[billing shipping].to_h do |name|
  sub_path = pubsub.subscription_path(name)
  sub_admin.create_subscription(name: sub_path, topic: topic_path, ack_deadline_seconds: 10)
  [name, sub_path]
end

# One Publish fans out to both subscriptions with a shared message id.
publisher = pubsub.publisher(topic.name)
msg_id = publisher.publish("order #1001 placed").message_id
puts "published: #{msg_id}"

# Each subscription has its own backlog.
sub_paths.each do |name, sub_path|
  subscriber = pubsub.subscriber(sub_path)
  rcv = subscriber.pull(immediate: false, max: 1).first
  puts "#{name} received message id=#{rcv.message_id}"
  rcv.acknowledge!
end

Filtered fan-out

Set a filter on a subscription so it receives only the publishes whose attributes match. Filters are an attributes-only CEL subset (attributes:KEY, = / !=, hasPrefix, AND / OR / NOT), ≤ 256 characters, compiled once at create-time, and applied at fan-out — a non-matching publish is never enqueued for that subscription, and a publish that matches no subscription still succeeds.

# A subscription that receives only "order" events.
subscriber.create_subscription(
    request={
        "name": subscriber.subscription_path(proj, "orders-only"),
        "topic": topic_path,
        "filter": 'attributes.eventType = "order"',
    }
)

# A matching publish is delivered; a non-matching one is suppressed for this subscription.
publisher.publish(topic_path, b"an order event", eventType="order")
publisher.publish(topic_path, b"a metric event", eventType="metric")  # suppressed for orders-only

The filter is immutable and attributes-only. It is compiled at CreateSubscription and cannot be changed afterward (UpdateSubscription rejects a filter change); a malformed filter is rejected at create-time with INVALID_ARGUMENT. Put the values you filter on into message attributes, not the body. See Message filtering.

Independent backlogs and ack state

Each subscription is a separate Queue channel gcp.sub.{subscription} with its own backlog, ack-deadline leases, dead-letter policy, and retention. A consumer that lags, nacks, or dead-letters on one subscription has no effect on any other subscription bound to the same topic — the topic log gcp.{topic} is the single shared, replayable source, and each subscription replays from it independently via Seek.

Fan-Out

Overview

How it works

Fan one publish to many subscriptions

Filtered fan-out

Independent backlogs and ack state

Publish & Subscribe

Message filtering

Channel mapping

On this page

Fan-Out

Overview

How it works

Fan one publish to many subscriptions

Filtered fan-out

Independent backlogs and ack state

Related

Publish & Subscribe

Message filtering

Channel mapping

On this page