Java函数如何利用发布/订阅模式提升代码可复用性？（函数.复用.订阅.提升.利用...）

wufei1232024-09-30java60

java函数如何利用发布/订阅模式提升代码可复用性？

利用 Java 函数和发布/订阅模式增强代码可复用性

在 Java 中，发布/订阅模式是一种设计模式，它允许多个事件消费者订阅事件发布者发布的事件。这种模式可以显著提高代码的可复用性，特别是对于事件驱动的系统。

实现

在 Java 中，可以使用 Google Cloud Pub/Sub 库来实现发布/订阅模式。该库提供了 Publisher 和 Subscriber 类来分别发布和订阅事件。

为了使用 Pub/Sub，首先需要创建一个项目并启用 Pub/Sub API。然后，可以使用以下步骤发布消息：

import com.google.api.client.util.Base64;
import com.google.cloud.pubsub.v1.Publisher;
import com.google.protobuf.ByteString;
import com.google.pubsub.v1.ProjectTopicName;
import com.google.pubsub.v1.PubsubMessage;
import java.io.IOException;

public class MessagePublisher {

  public static void main(String... args) throws Exception {
    String projectId = "your-project-id";
    String topicId = "your-topic-id";

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

    Publisher publisher = null;
    try {
      // Create a publisher instance with default settings bound to the topic
      publisher = Publisher.newBuilder(topicName).build();

      String message = "Hello World!";

      // Data must be a bytestring
      ByteString data = ByteString.copyFromUtf8(message);
      PubsubMessage pubsubMessage = PubsubMessage.newBuilder().setData(data).build();

      // Once published, returns a server-assigned message id (unique within the topic)
      String messageId = publisher.publish(pubsubMessage).get();
      System.out.println("Published a message with message id: " + messageId);

    } catch (IOException e) {
      System.out.println(e.toString());
    } finally {
      if (publisher != null) {
        // When finished with the publisher, shutdown to free up resources.
        publisher.shutdown();
        publisher.awaitTermination(1, TimeUnit.MINUTES);
      }
    }
  }
}

消息发布到主题后，可以由一个或多个订阅者接收。使用以下步骤订阅消息：

import com.google.cloud.pubsub.v1.AckReplyConsumer;
import com.google.cloud.pubsub.v1.MessageReceiver;
import com.google.cloud.pubsub.v1.Subscriber;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.PubsubMessage;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class MessageSubscriber {

  public static void main(String... args) throws Exception {
    String projectId = "your-project-id";
    String subscriptionId = "your-subscription-id";

    ProjectSubscriptionName subscriptionName =
        ProjectSubscriptionName.of(projectId, subscriptionId);

    Subscriber subscriber = null;
    try {
      subscriber = Subscriber.newBuilder(subscriptionName, new MessageReceiver() {
        @Override
        public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {
          // Handle incoming message, then ack the received message.
          System.out.println("Id: " + message.getMessageId());
          System.out.println("Data: " + message.getData().toStringUtf8());
          consumer.ack();
        }
      }).build();
      // Start the subscriber.
      subscriber.startAsync().awaitRunning();
      System.out.printf("Listening for messages on %s:\n", subscriptionName.toString());
      // Allow the subscriber to run for 30s unless an unrecoverable error occurs.
      subscriber.awaitTerminated(30, TimeUnit.SECONDS);
    } catch (TimeoutException timeoutException) {
      // Shut down the subscriber after 30s. Stop receiving messages.
      subscriber.stopAsync();
    }
  }
}

实战案例

以下是发布/订阅模式如何提高代码可复用性的一个实战案例：