创建和管理 Subscription

Subscription 的属性

  • ackTimeoutSeconds

    指定 HServer 将 records 标记为 unacked 的最大等待时间，之后该记录将被再次发送。

  • maxUnackedRecords。

    允许的未 acked record 的最大数量。超过设定的大小后，服务器将停止向相应的消费者 发送 records。

创建一个 Subscription

每个 subscription 都必须指定要订阅哪个 stream，这意味着你必须确保要订阅的 stream 已经被创建。

关于订阅的名称，请参考资源命名准则

当创建一个 subscription 时，你可以像这样提供提到的属性：

  1. // CreateSubscriptionExample.java
  3. package docs.code.examples;
  5. import io.hstream.HStreamClient;
  6. import io.hstream.Subscription;
  8. public class CreateSubscriptionExample {
  9.   public static void main(String[] args) throws Exception {
  10.     // TODO(developer): Replace these variables before running the sample.
  11.     String serviceUrl = "127.0.0.1:6570";
  12.     if (System.getenv("serviceUrl") != null) {
  13.       serviceUrl = System.getenv("serviceUrl");
  14.     }
  15.     String streamName = "your_h_records_stream_name";
  16.     HStreamClient client = HStreamClient.builder().serviceUrl(serviceUrl).build();
  17.     createSubscriptionExample(client, streamName);
  18.     client.close();
  19.   }
  21.   public static void createSubscriptionExample(HStreamClient client, String streamName) {
  22.     // TODO(developer): Specify the options while creating the subscription
  23.     String subscriptionId = "your_subscription_id";
  24.     Subscription subscription =
  25.         Subscription.newBuilder().subscription(subscriptionId).stream(streamName)
  26.             .ackTimeoutSeconds(600) // The default setting is 600 seconds
  27.             .maxUnackedRecords(10000) // The default setting is 10000 records
  28.             .build();
  29.     client.createSubscription(subscription);
  30.   }
  31. }
  1. // ExampleCreateSubscription.go
  3. package examples
  5. import (
  6.     "github.com/hstreamdb/hstreamdb-go/hstream"
  7.     "log"
  8. )
  10. func ExampleCreateSubscription() error {
  11.     client, err := hstream.NewHStreamClient(YourHStreamServiceUrl)
  12.     if err != nil {
  13.         log.Fatalf("Creating client error: %s", err)
  14.     }
  15.     defer client.Close()
  17.     streamName := "testStream"
  18.     subId0 := "SubscriptionId0"
  19.     subId1 := "SubscriptionId1"
  21.     // Create a new subscription with ack timeout = 60s, max unAcked records num set to 10000 and set
  22.     // subscriptionOffset to Earliest
  23.     if err = client.CreateSubscription(subId0, streamName,
  24.         hstream.WithAckTimeout(60),
  25.         hstream.WithMaxUnackedRecords(10000),
  26.         hstream.WithOffset(hstream.EARLIEST)); err != nil {
  27.         log.Fatalf("Creating subscription error: %s", err)
  28.     }
  30.     if err = client.CreateSubscription(subId1, streamName,
  31.         hstream.WithAckTimeout(600),
  32.         hstream.WithMaxUnackedRecords(5000),
  33.         hstream.WithOffset(hstream.LATEST)); err != nil {
  34.         log.Fatalf("Creating subscription error: %s", err)
  35.     }
  37.     return nil
  38. }
  1. # https://github.com/hstreamdb/hstreamdb-py/blob/main/examples/snippets/guides.py
  2. import asyncio
  3. import hstreamdb
  4. import os
  6. # NOTE: Replace with your own host and port
  7. host = os.getenv("GUIDE_HOST", "127.0.0.1")
  8. port = os.getenv("GUIDE_PORT", 6570)
  9. stream_name = "your_stream"
  10. subscription = "your_subscription"
  13. # Run: asyncio.run(main(your_async_function))
  14. async def main(*funcs):
  15.     async with await hstreamdb.insecure_client(host=host, port=port) as client:
  16.         for f in funcs:
  17.             await f(client)
  20. async def create_subscription(client):
  21.     await client.create_subscription(
  22.         subscription,
  23.         stream_name,
  24.         ack_timeout=600,
  25.         max_unacks=10000,
  26.         offset=hstreamdb.SpecialOffset.EARLIEST,
  27.     )

删除一个订阅

要删除一个的订阅，你需要确保没有活跃的订阅消费者，除非启用强制删除。

强制删除一个 Subscription

如果你确实想删除一个 subscription，并且有消费者正在运行，请启用强制删除。当强制 删除一个 subscription 时，该订阅将处于删除中的状态，并关闭正在运行的消费者，这意 味着你将无法加入、删除或创建一个同名的 subscription 。在删除完成后，你可以用同样 的名字创建一个订阅，这个订阅将是一个全新的订阅。即使他们订阅的是同一个流，这个新 的订阅也不会与被删除的订阅共享消费进度。

  1. // DeleteSubscriptionExample.java
  3. package docs.code.examples;
  5. import io.hstream.HStreamClient;
  7. public class DeleteSubscriptionExample {
  8.   public static void main(String[] args) throws Exception {
  9.     // TODO(developer): Replace these variables before running the sample.
  10.     // String serviceUrl = "your-service-url-address";
  11.     String serviceUrl = "127.0.0.1:6570";
  12.     if (System.getenv("serviceUrl") != null) {
  13.       serviceUrl = System.getenv("serviceUrl");
  14.     }
  16.     String subscriptionId = "your_subscription_id";
  17.     HStreamClient client = HStreamClient.builder().serviceUrl(serviceUrl).build();
  18.     deleteSubscriptionExample(client, subscriptionId);
  19.     client.close();
  20.   }
  22.   public static void deleteSubscriptionExample(HStreamClient client, String subscriptionId) {
  23.     client.deleteSubscription(subscriptionId);
  24.   }
  26.   public static void deleteSubscriptionForceExample(HStreamClient client, String subscriptionId) {
  27.     client.deleteSubscription(subscriptionId, true);
  28.   }
  29. }
  1. // ExampleDeleteSubscription.go
  3. package examples
  5. import (
  6.     "github.com/hstreamdb/hstreamdb-go/hstream"
  7.     "log"
  8. )
  10. func ExampleDeleteSubscription() error {
  11.     client, err := hstream.NewHStreamClient(YourHStreamServiceUrl)
  12.     if err != nil {
  13.         log.Fatalf("Creating client error: %s", err)
  14.     }
  15.     defer client.Close()
  17.     subId0 := "SubscriptionId0"
  18.     subId1 := "SubscriptionId1"
  20.     // force delete subscription
  21.     if err = client.DeleteSubscription(subId0, true); err != nil {
  22.         log.Fatalf("Force deleting subscription error: %s", err)
  23.     }
  25.     // delete subscription
  26.     if err = client.DeleteSubscription(subId1, false); err != nil {
  27.         log.Fatalf("Deleting subscription error: %s", err)
  28.     }
  30.     return nil
  31. }
  1. # https://github.com/hstreamdb/hstreamdb-py/blob/main/examples/snippets/guides.py
  2. import asyncio
  3. import hstreamdb
  4. import os
  6. # NOTE: Replace with your own host and port
  7. host = os.getenv("GUIDE_HOST", "127.0.0.1")
  8. port = os.getenv("GUIDE_PORT", 6570)
  9. stream_name = "your_stream"
  10. subscription = "your_subscription"
  13. # Run: asyncio.run(main(your_async_function))
  14. async def main(*funcs):
  15.     async with await hstreamdb.insecure_client(host=host, port=port) as client:
  16.         for f in funcs:
  17.             await f(client)
  20. async def delete_subscription(client):
  21.     await client.delete_subscription(subscription, force=True)

列出 HStream 中的 subscription 信息

  1. // ListSubscriptionsExample.java
  3. package docs.code.examples;
  5. import io.hstream.HStreamClient;
  6. import io.hstream.Subscription;
  7. import java.util.List;
  9. public class ListSubscriptionsExample {
  11.   public static void main(String[] args) throws Exception {
  12.     // TODO(developer): Replace these variables before running the sample.
  13.     String serviceUrl = "127.0.0.1:6570";
  14.     if (System.getenv("serviceUrl") != null) {
  15.       serviceUrl = System.getenv("serviceUrl");
  16.     }
  18.     HStreamClient client = HStreamClient.builder().serviceUrl(serviceUrl).build();
  19.     listSubscriptionExample(client);
  20.     client.close();
  21.   }
  23.   public static void listSubscriptionExample(HStreamClient client) {
  24.     List<Subscription> subscriptions = client.listSubscriptions();
  25.     for (Subscription subscription : subscriptions) {
  26.       System.out.println(subscription.getSubscriptionId());
  27.     }
  28.   }
  29. }
  1. // ExampleListSubscriptions.go
  3. package examples
  5. import (
  6.     "fmt"
  7.     "github.com/hstreamdb/hstreamdb-go/hstream"
  8.     "log"
  9. )
  11. func ExampleListSubscriptions() error {
  12.     client, err := hstream.NewHStreamClient(YourHStreamServiceUrl)
  13.     if err != nil {
  14.         log.Fatalf("Creating client error: %s", err)
  15.     }
  16.     defer client.Close()
  18.     subscriptions, err := client.ListSubscriptions()
  19.     if err != nil {
  20.         log.Fatalf("Listing subscriptions error: %s", err)
  21.     }
  23.     for _, sub := range subscriptions {
  24.         fmt.Printf("%+v\n", sub)
  25.     }
  27.     return nil
  28. }
  1. # https://github.com/hstreamdb/hstreamdb-py/blob/main/examples/snippets/guides.py
  2. import asyncio
  3. import hstreamdb
  4. import os
  6. # NOTE: Replace with your own host and port
  7. host = os.getenv("GUIDE_HOST", "127.0.0.1")
  8. port = os.getenv("GUIDE_PORT", 6570)
  9. stream_name = "your_stream"
  10. subscription = "your_subscription"
  13. # Run: asyncio.run(main(your_async_function))
  14. async def main(*funcs):
  15.     async with await hstreamdb.insecure_client(host=host, port=port) as client:
  16.         for f in funcs:
  17.             await f(client)
  20. async def list_subscriptions(client):
  21.     subscriptions = await client.list_subscriptions()
  22.     for s in subscriptions:
  23.         print(s)