Google Cloud Pub/Sub

Giriş

Açıklaması şöyle. Sanırım "AWS Simple Queue Service" ile aynı şey.

Pub/Sub is a messaging solution provided by GCP.

Google Cloud Pub/Sub vs Kafka

Google Cloud Pub/Sub at-least-once messaging semantics sağlar. 

Kullanım

Açıklaması şöyle

1. Create a new project in the Google Cloud Console.

2. Enable the Pub/Sub API for your project.

3. Create a new topic.

4. Create a new subscription to that topic.

5. Download a private key file for your service account.

Örnek - SpringBoot

application.properties dosyasında şöyle yaparız

spring.cloud.gcp.project-id=your-gcp-project-id
spring.cloud.gcp.pubsub.credentials.location=file:path/to/your/service/account/key.json

Pusblisher

Şöyle yaparız

import org.springframework.cloud.gcp.pubsub.core.PubSubTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class PublisherController {
    private final PubSubTemplate pubSubTemplate;
    public PublisherController(PubSubTemplate pubSubTemplate) {
        this.pubSubTemplate = pubSubTemplate;
    }
    @PostMapping("/publish")
    public String publishMessage(@RequestParam("message") String message) {
        pubSubTemplate.publish("my-topic", message);
        return "Message published successfully.";
    }
}

Subscriber

Şöyle yaparız

import com.google.cloud.pubsub.v1.AckReplyConsumer;
import com.google.pubsub.v1.PubsubMessage;
import org.springframework.cloud.gcp.pubsub.support.GcpPubSubHeaders;
import org.springframework.cloud.gcp.pubsub.support.BasicAcknowledgeablePubsubMessage;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.cloud.gcp.pubsub.integration.AckMode;
import org.springframework.cloud.gcp.pubsub.integration.inbound.PubSubInboundChannelAdapter;
import org.springframework.cloud.gcp.pubsub.integration.inbound.PubSubMessageSource;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.annotation.Poller;
import org.springframework.stereotype.Service;

@Service
public class SubscriberService {

  @ServiceActivator(inputChannel = "mySubscriptionInputChannel")
  public void messageReceiver(String payload, 
                              @Header(GcpPubSubHeaders.ORIGINAL_MESSAGE) 
                              BasicAcknowledgeablePubsubMessage message) {
    System.out.println("Message received from Google Cloud Pub/Sub: " + payload);

    // Acknowledge the message upon successful processing
    message.ack();
  }
}

Configure the input channel

Şöyle yaparız

import org.springframework.cloud.gcp.pubsub.integration.inbound.PubSubInboundChannelAdapter;
import org.springframework.cloud.gcp.pubsub.support.converter.SimplePubSubMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.messaging.MessageChannel;

@Configuration
public class PubSubConfig {
  @Bean
  public MessageChannel myInputChannel() {
    return new DirectChannel();
  }
  @Bean
  public PubSubInboundChannelAdapter messageChannelAdapter(
    @Qualifier("myInputChannel") MessageChannel inputChannel,
    PubSubTemplate pubSubTemplate) {
    PubSubInboundChannelAdapter adapter = new PubSubInboundChannelAdapter(pubSubTemplate,
      "my-subscription");
    adapter.setOutputChannel(inputChannel);
    adapter.setPayloadConverter(new SimplePubSubMessageConverter());
    return adapter;
  }
}

Açıklaması şöyle

In the above configuration, we create a PubSubInboundChannelAdapter that listens to the "my-subscription" subscription and forwards the incoming messages to the "myInputChannel" channel.