为了做到这一点，你需要

首先关闭pubsub的GCP自动配置

@SpringBootApplication(exclude = {
        GcpPubSubAutoConfiguration.class,
        GcpPubSubReactiveAutoConfiguration.class
})
public class PubsubApplication {

    public static void main(String[] args) {
        SpringApplication.run(PubsubApplication.class, args);
    }

}

然后为第一个项目创建配置

@Configuration
public class Project1Config {

  private static final Logger LOGGER = LoggerFactory.getLogger(Project1Config.class);

  @Bean(name = "project1_IdProvider")
  public GcpProjectIdProvider project1_IdProvider() {
    return new DefaultGcpProjectIdProvider() {
      @Override
      public String getProjectId() {
        return "YOURPROJECTID";
      }
    };
  }

  @Bean(name = "project1_credentialsProvider")
  public CredentialsProvider project1_credentialsProvider() throws IOException {
    return new CredentialsProvider() {
      @Override
      public Credentials getCredentials() throws IOException {
        return ServiceAccountCredentials.fromStream(
            new ClassPathResource("YOURCREDENTIALS").getInputStream());
      }
    };
  }

  @Bean("project1_pubSubSubscriberTemplate")
  public PubSubSubscriberTemplate pubSubSubscriberTemplate(
          @Qualifier("project1_subscriberFactory") SubscriberFactory subscriberFactory) {
    return new PubSubSubscriberTemplate(subscriberFactory);
  }


  @Bean("project1_publisherFactory")
  public DefaultPublisherFactory publisherFactory(
          @Qualifier("project1_IdProvider") GcpProjectIdProvider projectIdProvider,
          @Qualifier("project1_credentialsProvider") CredentialsProvider credentialsProvider) {
    final DefaultPublisherFactory defaultPublisherFactory = new DefaultPublisherFactory(projectIdProvider);
    defaultPublisherFactory.setCredentialsProvider(credentialsProvider);
    return defaultPublisherFactory;
  }

  @Bean("project1_subscriberFactory")
  public DefaultSubscriberFactory subscriberFactory(
          @Qualifier("project1_IdProvider") GcpProjectIdProvider projectIdProvider,
          @Qualifier("project1_credentialsProvider") CredentialsProvider credentialsProvider) {
    final DefaultSubscriberFactory defaultSubscriberFactory = new DefaultSubscriberFactory(projectIdProvider);
    defaultSubscriberFactory.setCredentialsProvider(credentialsProvider);
    return defaultSubscriberFactory;
  }

  @Bean(name = "project1_pubsubInputChannel")
  public MessageChannel pubsubInputChannel() {
    return new DirectChannel();
  }

  @Bean(name = "project1_pubSubTemplate")
  public PubSubTemplate project1_PubSubTemplate(
      @Qualifier("project1_publisherFactory") PublisherFactory publisherFactory,
      @Qualifier("project1_subscriberFactory") SubscriberFactory subscriberFactory,
      @Qualifier("project1_credentialsProvider") CredentialsProvider credentialsProvider) {
    if (publisherFactory instanceof DefaultPublisherFactory) {
      ((DefaultPublisherFactory) publisherFactory).setCredentialsProvider(credentialsProvider);
    }
    return new PubSubTemplate(publisherFactory, subscriberFactory);
  }

  @Bean(name = "project1_messageChannelAdapter")
  public PubSubInboundChannelAdapter messageChannelAdapter(
      @Qualifier("project1_pubsubInputChannel") MessageChannel inputChannel,
      @Qualifier("project1_pubSubTemplate") PubSubTemplate pubSubTemplate) {

    PubSubInboundChannelAdapter adapter =
        new PubSubInboundChannelAdapter(pubSubTemplate, "YOURSUBSCRIPTIONNAME");
    adapter.setOutputChannel(inputChannel);
    adapter.setAckMode(AckMode.MANUAL);
    return adapter;
  }

  @Bean("project1_messageReceiver")
  @ServiceActivator(inputChannel = "project1_pubsubInputChannel")
  public MessageHandler messageReceiver() {
    return message -> {
      LOGGER.info("Message arrived! Payload: " + new String((byte[]) message.getPayload()));
      LOGGER.info("Message headers {}", message.getHeaders());
      BasicAcknowledgeablePubsubMessage originalMessage =
          message
              .getHeaders()
              .get(GcpPubSubHeaders.ORIGINAL_MESSAGE, BasicAcknowledgeablePubsubMessage.class);
      originalMessage.ack();
    };
  }

  @Bean("project1_messageSender")
  @ServiceActivator(inputChannel = "project1_pubsubOutputChannel")
  public MessageHandler messageSender(
          @Qualifier("project1_pubSubTemplate") PubSubTemplate pubsubTemplate) {
    return new PubSubMessageHandler(pubsubTemplate, "YOURTOPICNAME");
  }
}

Next-为project ect2创建配置

@Configuration
public class Project2Config {

  private static final Logger LOGGER = LoggerFactory.getLogger(Project2Config.class);

  @Bean(name = "project2_IdProvider")
  public DefaultGcpProjectIdProvider project2_IdProvider() {
    return new DefaultGcpProjectIdProvider() {
      @Override
      public String getProjectId() {
        return "project-id-lksjfkalsdjfkl";
      }
    };
  }

  @Bean(name = "project2_credentialsProvider")
  public CredentialsProvider project2_credentialsProvider() throws IOException {
    return new CredentialsProvider() {
      @Override
      public Credentials getCredentials() throws IOException {
        return ServiceAccountCredentials.fromStream(
            new ClassPathResource("project2.json").getInputStream());
      }
    };
  }

  @Bean("project2_pubSubSubscriberTemplate")
  public PubSubSubscriberTemplate pubSubSubscriberTemplate(
          @Qualifier("project2_subscriberFactory") SubscriberFactory subscriberFactory) {
    return new PubSubSubscriberTemplate(subscriberFactory);
  }

  @Bean("project2_publisherFactory")
  public DefaultPublisherFactory publisherFactory(
          @Qualifier("project2_IdProvider") GcpProjectIdProvider projectIdProvider,
          @Qualifier("project2_credentialsProvider") CredentialsProvider credentialsProvider) {
    final DefaultPublisherFactory defaultPublisherFactory = new DefaultPublisherFactory(projectIdProvider);
    defaultPublisherFactory.setCredentialsProvider(credentialsProvider);
    return defaultPublisherFactory;
  }

  @Bean("project2_subscriberFactory")
  public DefaultSubscriberFactory subscriberFactory(
          @Qualifier("project2_IdProvider") GcpProjectIdProvider projectIdProvider,
          @Qualifier("project2_credentialsProvider") CredentialsProvider credentialsProvider) {
    final DefaultSubscriberFactory defaultSubscriberFactory = new DefaultSubscriberFactory(projectIdProvider);
    defaultSubscriberFactory.setCredentialsProvider(credentialsProvider);
    return defaultSubscriberFactory;
  }

  @Bean(name = "project2_pubsubInputChannel")
  public MessageChannel pubsubInputChannel() {
    return new DirectChannel();
  }

  @Bean(name = "project2_pubSubTemplate")
  public PubSubTemplate project2_PubSubTemplate(
      @Qualifier("project2_publisherFactory") PublisherFactory publisherFactory,
      @Qualifier("project2_subscriberFactory") SubscriberFactory subscriberFactory,
      @Qualifier("project2_credentialsProvider") CredentialsProvider credentialsProvider) {
    if (publisherFactory instanceof DefaultPublisherFactory) {
      ((DefaultPublisherFactory) publisherFactory).setCredentialsProvider(credentialsProvider);
    }
    return new PubSubTemplate(publisherFactory, subscriberFactory);
  }

  @Bean(name = "project2_messageChannelAdapter")
  public PubSubInboundChannelAdapter messageChannelAdapter(
      @Qualifier("project2_pubsubInputChannel") MessageChannel inputChannel,
      @Qualifier("project2_pubSubTemplate") PubSubTemplate pubSubTemplate) {

    PubSubInboundChannelAdapter adapter =
        new PubSubInboundChannelAdapter(pubSubTemplate, "project2-testSubscription");
    adapter.setOutputChannel(inputChannel);
    adapter.setAckMode(AckMode.MANUAL);
    return adapter;
  }

  @Bean("project2_messageReceiver")
  @ServiceActivator(inputChannel = "project2_pubsubInputChannel")
  public MessageHandler messageReceiver() {
    return message -> {
      LOGGER.info("Message Payload: " + new String((byte[]) message.getPayload()));
      LOGGER.info("Message headers {}", message.getHeaders());
      BasicAcknowledgeablePubsubMessage originalMessage =
          message
              .getHeaders()
              .get(GcpPubSubHeaders.ORIGINAL_MESSAGE, BasicAcknowledgeablePubsubMessage.class);
      originalMessage.ack();
    };
  }

  @Bean(name = "project2_messageSender")
  @ServiceActivator(inputChannel = "project2_pubsubOutputChannel")
  public MessageHandler messageSender(
          @Qualifier("project2_pubSubTemplate") PubSubTemplate pubsubTemplate) {
    return new PubSubMessageHandler(pubsubTemplate, "project2-testTopic");
  }
}

为项目1创建出站网关

project1_pubsubOutputChannel -在Project1Config中指定

@Service
@MessagingGateway(defaultRequestChannel = "project1_pubsubOutputChannel")
public interface Project1PubsubOutboundGateway {

  void sendToPubsub(String text);
}

为项目2创建出站网关

project2_pubsubOutputChannel-在Project2Config中指定

@Service
@MessagingGateway(defaultRequestChannel = "project2_pubsubOutputChannel")
public interface Project2PubsubOutboundGateway {

  void sendToPubsub(String text);
}

现在我们成功了：

@RestController
public class WebAppController {

  // tag::autowireGateway[]
  @Autowired private Project1PubsubOutboundGateway project1PubsubOutboundGateway;
  @Autowired private Project2PubsubOutboundGateway project2PubsubOutboundGateway;
  // end::autowireGateway[]

  @PostMapping("/publishMessage")
  public ResponseEntity<String> publishMessage(@RequestParam("message") String message) {
    project1PubsubOutboundGateway.sendToPubsub(message);
    project2PubsubOutboundGateway.sendToPubsub(message);
    return ResponseEntity.ok("OK");
  }
}

检查日志以查看消息传递是否正常

查看git项目的更多细节:https://github.com/olgmaks/spring-gcppubsub-multiproject

我有相同类型的需求，我想从另一个GCP项目获取浏览器数据，从不同的项目获取应用程序数据。

运行此命令的先决条件是您需要一个服务帐户，该帐户应有权从两个项目中获取数据。

假设 defaultProjectId 是 project1，insightsProjectId 是 application.properties 中的 project2，我们将使用@Value注释获取值。

这在 Pubsubconfig 类中对我有用.java该类是 pubsub 的配置豆，我能够从两个不同的项目中读取多个订阅

下面是代码

PubSubSubscriberTemplate returnDefaultProject() {
    DefaultSubscriberFactory defaultFactory = new DefaultSubscriberFactory(() -> defaultProjectId);
    return new PubSubSubscriberTemplate(defaultFactory);
}

PubSubSubscriberTemplate returnInsightsProject() {
    DefaultSubscriberFactory insightsFactory = new DefaultSubscriberFactory(() -> insightsProjectId);
    return new PubSubSubscriberTemplate(insightsFactory);
}

@Bean(name = "browserChannelAdapter")
public PubSubInboundChannelAdapter browserChannelAdapter(
        @Qualifier("browserInputChannel") MessageChannel inputChannel) {
    PubSubInboundChannelAdapter adapter =
            new PubSubInboundChannelAdapter(returnInsightsProject(), brSubscriptionId);
    adapter.setOutputChannel(inputChannel);

    return adapter;
}

@Bean(name = "appChannelAdapter")
public PubSubInboundChannelAdapter appChannelAdapter(
        @Qualifier("appInputChannel") MessageChannel inputChannel) {
    PubSubInboundChannelAdapter adapter =
            new PubSubInboundChannelAdapter(returnDefaultProject(), appSubscriptionId);
    adapter.setOutputChannel(inputChannel);

    return adapter;
}