创建和管理 Subscription

创建和管理 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)