This is the multi-page printable view of this section. Click here to print.

Return to the regular view of this page.

Dapr Java SDK

用于开发 Dapr 应用程序的 Java SDK 软件包

Dapr 提供了多种软件包来协助 Java 应用程序的开发。使用它们,你可以通过 Dapr 创建 Java 客户端、服务器和虚拟 Actor。

前提条件

导入 Dapr Java SDK

接下来,导入 Java SDK 软件包以开始使用。选择你首选的构建工具以了解如何导入。

对于 Maven 项目,将以下内容添加到你的 pom.xml 文件中:

<project>
  ...
  <dependencies>
    ...
    <!-- Dapr's core SDK with all features, except Actors. -->
    <dependency>
      <groupId>io.dapr</groupId>
      <artifactId>dapr-sdk</artifactId>
      <version>1.16.0</version>
    </dependency>
    <!-- Dapr's SDK for Actors (optional). -->
    <dependency>
      <groupId>io.dapr</groupId>
      <artifactId>dapr-sdk-actors</artifactId>
      <version>1.16.0</version>
    </dependency>
    <!-- Dapr's SDK integration with SpringBoot (optional). -->
    <dependency>
      <groupId>io.dapr</groupId>
      <artifactId>dapr-sdk-springboot</artifactId>
      <version>1.16.0</version>
    </dependency>
    ...
  </dependencies>
  ...
</project>

对于 Gradle 项目,将以下内容添加到你的 build.gradle 文件中:

dependencies {
...
    // Dapr's core SDK with all features, except Actors.
    compile('io.dapr:dapr-sdk:1.16.0')
    // Dapr's SDK for Actors (optional).
    compile('io.dapr:dapr-sdk-actors:1.16.0')
    // Dapr's SDK integration with SpringBoot (optional).
    compile('io.dapr:dapr-sdk-springboot:1.16.0')
}

如果你还使用了 Spring Boot,可能会遇到一个常见问题:Dapr SDK 使用的 OkHttp 版本与 Spring Boot 的 Bill of Materials 中指定的版本冲突。

你可以通过在项目中指定与 Dapr SDK 使用的版本兼容的 OkHttp 版本来解决此问题:

<dependency>
  <groupId>com.squareup.okhttp3</groupId>
  <artifactId>okhttp</artifactId>
  <version>1.16.0</version>
</dependency>

试用

测试 Dapr Java SDK。通过 Java 快速入门和教程来查看 Dapr 的实际效果:

SDK 示例描述
快速入门在几分钟内使用 Java SDK 体验 Dapr 的 API 构建块。
SDK 示例克隆 SDK 仓库以尝试一些示例并开始使用。
import io.dapr.client.DaprClient;
import io.dapr.client.DaprClientBuilder;

try (DaprClient client = (new DaprClientBuilder()).build()) {
  // sending a class with message; BINDING_OPERATION="create"
  client.invokeBinding(BINDING_NAME, BINDING_OPERATION, myClass).block();

  // sending a plain string
  client.invokeBinding(BINDING_NAME, BINDING_OPERATION, message).block();
}

可用软件包

客户端

创建与 Dapr 边车和其他 Dapr 应用程序交互的 Java 客户端。

工作流

在 Java 中创建和管理与其他 Dapr API 协同工作的 workflow。

1 - AI

借助 Dapr Conversation AI 包,您可以从 Java 应用程序与 Dapr AI 工作负载进行交互。要开始使用,请参阅 Dapr AI 操作指南。

1.1 - 操作指南:使用 Java SDK 编写和管理 Dapr Conversation AI

如何使用 Dapr Java SDK 快速上手 Conversation AI

在本次演示中,我们将介绍如何使用 Conversation API 与大语言模型(LLM)进行对话。该 API 会返回给定提示词的 LLM 响应。通过提供的 conversation ai 示例,你将:

此示例使用自托管模式dapr init 的默认配置。

前置条件

设置环境

克隆 Java SDK 仓库 并进入该目录。

git clone https://github.com/dapr/java-sdk.git
cd java-sdk

运行以下命令以安装使用 Dapr Java SDK 运行 Conversation AI 示例所需的依赖。

mvn clean install -DskipTests

从 Java SDK 根目录进入示例目录。

cd examples

运行 Dapr 边车。

dapr run --app-id conversationapp --dapr-grpc-port 51439 --dapr-http-port 3500 --app-port 8080

现在,Dapr 正在 http://localhost:3500 监听 HTTP 请求,在 http://localhost:51439 监听 gRPC 请求。

向 Conversation AI API 发送包含个人身份信息(PII)的提示词

DemoConversationAI 中,包含使用 DaprPreviewClient 下的 converse 方法发送提示词的步骤。

public class DemoConversationAI {
  /**
   * 启动客户端的 main 方法。
   *
   * @param args 输入参数(未使用)。
   */
  public static void main(String[] args) {
    try (DaprPreviewClient client = new DaprClientBuilder().buildPreviewClient()) {
      System.out.println("Sending the following input to LLM: Hello How are you? This is the my number 672-123-4567");

      ConversationInput daprConversationInput = new ConversationInput("Hello How are you? "
              + "This is the my number 672-123-4567");

      // 组件名称是在 conversation.yaml 文件的 metadata 块中提供的名称。
      Mono<ConversationResponse> responseMono = client.converse(new ConversationRequest("echo",
              List.of(daprConversationInput))
              .setContextId("contextId")
              .setScrubPii(true).setTemperature(1.1d));
      ConversationResponse response = responseMono.block();
      System.out.printf("Conversation output: %s", response.getConversationOutputs().get(0).getResult());
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
  }
}

使用以下命令运行 DemoConversationAI

java -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.conversation.DemoConversationAI

示例输出

== APP == Conversation output: Hello How are you? This is the my number <ISBN>

如输出所示,发送到 API 的号码已被混淆并以 <ISBN> 的形式返回。 上面的示例使用了一个“echo” 组件进行测试,该组件只是简单地返回输入消息。 当与 OpenAI 或 Claude 等 LLM 集成时,你将收到有意义的响应,而不是回显的输入。

后续步骤

2 - Dapr 客户端 Java SDK 入门

如何开始使用 Dapr Java SDK

Dapr 客户端包允许您从 Java 应用程序与其他 Dapr 应用程序进行交互。

前置条件

完成初始设置并将 Java SDK 导入您的项目

初始化客户端

您可以像这样初始化 Dapr 客户端:

DaprClient client = new DaprClientBuilder().build()

这将连接到默认的 Dapr gRPC 端点 localhost:50001。有关使用环境变量和系统属性配置客户端的信息,请参阅属性

错误处理

最初,Dapr 中的错误遵循标准 gRPC 错误模型。然而,为了提供更详细和有信息量的错误消息,在 1.13 版本中引入了增强的错误模型,该模型与 gRPC 更丰富的错误模型保持一致。作为响应,Java SDK 扩展了 DaprException 以包含 Dapr 中添加的错误详细信息。

使用 Dapr Java SDK 时处理 DaprException 并使用错误详细信息的示例:

...
      try {
        client.publishEvent("unknown_pubsub", "mytopic", "mydata").block();
      } catch (DaprException exception) {
        System.out.println("Dapr exception's error code: " + exception.getErrorCode());
        System.out.println("Dapr exception's message: " + exception.getMessage());
        // DaprException 现在包含 `getStatusDetails()` 以包含有关 Dapr 运行时错误的更多详细信息。
        System.out.println("Dapr exception's reason: " + exception.getStatusDetails().get(
        DaprErrorDetails.ErrorDetailType.ERROR_INFO,
            "reason",
        TypeRef.STRING));
      }
...

构建块

Java SDK 允许您与所有 Dapr 构建块 进行交互。

调用服务

import io.dapr.client.DaprClient;
import io.dapr.client.DaprClientBuilder;

try (DaprClient client = (new DaprClientBuilder()).build()) {
  // 调用 'GET' 方法 (HTTP),跳过序列化:使用 Mono<byte[]> 返回类型
  // 对于 gRPC,请在下面设置 HttpExtension.NONE 参数
  response = client.invokeMethod(SERVICE_TO_INVOKE, METHOD_TO_INVOKE, "{\"name\":\"World!\"}", HttpExtension.GET, byte[].class).block();

  // 调用 'POST' 方法 (HTTP),跳过序列化:使用 Mono<byte[]> 返回类型     
  response = client.invokeMethod(SERVICE_TO_INVOKE, METHOD_TO_INVOKE, "{\"id\":\"100\", \"FirstName\":\"Value\", \"LastName\":\"Value\"}", HttpExtension.POST, byte[].class).block();

  System.out.println(new String(response));

  // 调用 'POST' 方法 (HTTP),使用序列化:使用 Mono<Employee> 返回类型      
  Employee newEmployee = new Employee("Nigel", "Guitarist");
  Employee employeeResponse = client.invokeMethod(SERVICE_TO_INVOKE, "employees", newEmployee, HttpExtension.POST, Employee.class).block();
}

保存和获取应用程序状态

import io.dapr.client.DaprClient;
import io.dapr.client.DaprClientBuilder;
import io.dapr.client.domain.State;
import reactor.core.publisher.Mono;

try (DaprClient client = (new DaprClientBuilder()).build()) {
  // 保存状态
  client.saveState(STATE_STORE_NAME, FIRST_KEY_NAME, myClass).block();

  // 获取状态
  State<MyClass> retrievedMessage = client.getState(STATE_STORE_NAME, FIRST_KEY_NAME, MyClass.class).block();

  // 删除状态
  client.deleteState(STATE_STORE_NAME, FIRST_KEY_NAME).block();
}

发布和订阅消息

发布消息
import io.dapr.client.DaprClient;
import io.dapr.client.DaprClientBuilder;
import io.dapr.client.domain.Metadata;
import static java.util.Collections.singletonMap;

try (DaprClient client = (new DaprClientBuilder()).build()) {
  client.publishEvent(PUBSUB_NAME, TOPIC_NAME, message, singletonMap(Metadata.TTL_IN_SECONDS, MESSAGE_TTL_IN_SECONDS)).block();
}
订阅消息
import com.fasterxml.jackson.databind.ObjectMapper;
import io.dapr.Topic;
import io.dapr.client.domain.BulkSubscribeAppResponse;
import io.dapr.client.domain.BulkSubscribeAppResponseEntry;
import io.dapr.client.domain.BulkSubscribeAppResponseStatus;
import io.dapr.client.domain.BulkSubscribeMessage;
import io.dapr.client.domain.BulkSubscribeMessageEntry;
import io.dapr.client.domain.CloudEvent;
import io.dapr.springboot.annotations.BulkSubscribe;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Mono;

@RestController
public class SubscriberController {

  private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();

  @Topic(name = "testingtopic", pubsubName = "${myAppProperty:messagebus}")
  @PostMapping(path = "/testingtopic")
  public Mono<Void> handleMessage(@RequestBody(required = false) CloudEvent<?> cloudEvent) {
    return Mono.fromRunnable(() -> {
      try {
        System.out.println("Subscriber got: " + cloudEvent.getData());
        System.out.println("Subscriber got: " + OBJECT_MAPPER.writeValueAsString(cloudEvent));
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    });
  }

  @Topic(name = "testingtopic", pubsubName = "${myAppProperty:messagebus}",
          rule = @Rule(match = "event.type == 'myevent.v2'", priority = 1))
  @PostMapping(path = "/testingtopicV2")
  public Mono<Void> handleMessageV2(@RequestBody(required = false) CloudEvent envelope) {
    return Mono.fromRunnable(() -> {
      try {
        System.out.println("Subscriber got: " + cloudEvent.getData());
        System.out.println("Subscriber got: " + OBJECT_MAPPER.writeValueAsString(cloudEvent));
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    });
  }

  @BulkSubscribe()
  @Topic(name = "testingtopicbulk", pubsubName = "${myAppProperty:messagebus}")
  @PostMapping(path = "/testingtopicbulk")
  public Mono<BulkSubscribeAppResponse> handleBulkMessage(
          @RequestBody(required = false) BulkSubscribeMessage<CloudEvent<String>> bulkMessage) {
    return Mono.fromCallable(() -> {
      if (bulkMessage.getEntries().size() == 0) {
        return new BulkSubscribeAppResponse(new ArrayList<BulkSubscribeAppResponseEntry>());
      }

      System.out.println("Bulk Subscriber received " + bulkMessage.getEntries().size() + " messages.");

      List<BulkSubscribeAppResponseEntry> entries = new ArrayList<BulkSubscribeAppResponseEntry>();
      for (BulkSubscribeMessageEntry<?> entry : bulkMessage.getEntries()) {
        try {
          System.out.printf("Bulk Subscriber message has entry ID: %s\n", entry.getEntryId());
          CloudEvent<?> cloudEvent = (CloudEvent<?>) entry.getEvent();
          System.out.printf("Bulk Subscriber got: %s\n", cloudEvent.getData());
          entries.add(new BulkSubscribeAppResponseEntry(entry.getEntryId(), BulkSubscribeAppResponseStatus.SUCCESS));
        } catch (Exception e) {
          e.printStackTrace();
          entries.add(new BulkSubscribeAppResponseEntry(entry.getEntryId(), BulkSubscribeAppResponseStatus.RETRY));
        }
      }
      return new BulkSubscribeAppResponse(entries);
    });
  }
}
批量发布消息
import io.dapr.client.DaprClient;
import io.dapr.client.DaprClientBuilder;
import io.dapr.client.domain.BulkPublishResponse;
import io.dapr.client.domain.BulkPublishResponseFailedEntry;
import java.util.ArrayList;
import java.util.List;
class Solution {
  public void publishMessages() {
    try (DaprClient client = (new DaprClientBuilder()).build()) {
      // 创建要发布的消息列表
      List<String> messages = new ArrayList<>();
      for (int i = 0; i < NUM_MESSAGES; i++) {
        String message = String.format("This is message #%d", i);
        messages.add(message);
        System.out.println("Going to publish message : " + message);
      }

      // 使用批量发布 API 发布消息列表
      BulkPublishResponse<String> res = client.publishEvents(PUBSUB_NAME, TOPIC_NAME, "text/plain", messages).block()
    }
  }
}

与输出绑定交互

import io.dapr.client.DaprClient;
import io.dapr.client.DaprClientBuilder;

try (DaprClient client = (new DaprClientBuilder()).build()) {
  // 使用消息发送类;BINDING_OPERATION="create"
  client.invokeBinding(BINDING_NAME, BINDING_OPERATION, myClass).block();

  // 发送纯字符串
  client.invokeBinding(BINDING_NAME, BINDING_OPERATION, message).block();
}

与输入绑定交互

import org.springframework.web.bind.annotation.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@RestController
@RequestMapping("/")
public class myClass {
    private static final Logger log = LoggerFactory.getLogger(myClass);
        @PostMapping(path = "/checkout")
        public Mono<String> getCheckout(@RequestBody(required = false) byte[] body) {
            return Mono.fromRunnable(() ->
                    log.info("Received Message: " + new String(body)));
        }
}

获取密钥

import com.fasterxml.jackson.databind.ObjectMapper;
import io.dapr.client.DaprClient;
import io.dapr.client.DaprClientBuilder;
import java.util.Map;

try (DaprClient client = (new DaprClientBuilder()).build()) {
  Map<String, String> secret = client.getSecret(SECRET_STORE_NAME, secretKey).block();
  System.out.println(JSON_SERIALIZER.writeValueAsString(secret));
}

Actor

Actor 是一个隔离的、独立的计算和状态单元,具有单线程执行。Dapr 提供基于虚拟 Actor 模式的 actor 实现,该模式提供单线程编程模型,其中 actor 在不使用时会被垃圾回收。使用 Dapr 的实现,您可以根据 Actor 模型编写 Dapr actor,Dapr 利用底层平台提供的可扩展性和可靠性。

import io.dapr.actors.ActorMethod;
import io.dapr.actors.ActorType;
import reactor.core.publisher.Mono;

@ActorType(name = "DemoActor")
public interface DemoActor {

  void registerReminder();

  @ActorMethod(name = "echo_message")
  String say(String something);

  void clock(String message);

  @ActorMethod(returns = Integer.class)
  Mono<Integer> incrementAndGet(int delta);
}

获取和订阅应用程序配置

请注意,这是一个预览 API,因此只能通过 DaprPreviewClient 接口访问,而不能通过普通的 DaprClient 接口访问

import io.dapr.client.DaprClientBuilder;
import io.dapr.client.DaprPreviewClient;
import io.dapr.client.domain.ConfigurationItem;
import io.dapr.client.domain.GetConfigurationRequest;
import io.dapr.client.domain.SubscribeConfigurationRequest;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

try (DaprPreviewClient client = (new DaprClientBuilder()).buildPreviewClient()) {
  // 获取单个键的配置
  Mono<ConfigurationItem> item = client.getConfiguration(CONFIG_STORE_NAME, CONFIG_KEY).block();

  // 获取多个键的配置
  Mono<Map<String, ConfigurationItem>> items =
          client.getConfiguration(CONFIG_STORE_NAME, CONFIG_KEY_1, CONFIG_KEY_2);

  // 订阅配置更改
  Flux<SubscribeConfigurationResponse> outFlux = client.subscribeConfiguration(CONFIG_STORE_NAME, CONFIG_KEY_1, CONFIG_KEY_2);
  outFlux.subscribe(configItems -> configItems.forEach(...));

  // 取消订阅配置更改
  Mono<UnsubscribeConfigurationResponse> unsubscribe = client.unsubscribeConfiguration(SUBSCRIPTION_ID, CONFIG_STORE_NAME)
}

查询已保存的状态

请注意,这是一个预览 API,因此只能通过 DaprPreviewClient 接口访问,而不能通过普通的 DaprClient 接口访问

import io.dapr.client.DaprClient;
import io.dapr.client.DaprClientBuilder;
import io.dapr.client.DaprPreviewClient;
import io.dapr.client.domain.QueryStateItem;
import io.dapr.client.domain.QueryStateRequest;
import io.dapr.client.domain.QueryStateResponse;
import io.dapr.client.domain.query.Query;
import io.dapr.client.domain.query.Sorting;
import io.dapr.client.domain.query.filters.EqFilter;

try (DaprClient client = builder.build(); DaprPreviewClient previewClient = builder.buildPreviewClient()) {
        String searchVal = args.length == 0 ? "searchValue" : args[0];
        
        // 创建 JSON 数据
        Listing first = new Listing();
        first.setPropertyType("apartment");
        first.setId("1000");
        ...
        Listing second = new Listing();
        second.setPropertyType("row-house");
        second.setId("1002");
        ...
        Listing third = new Listing();
        third.setPropertyType("apartment");
        third.setId("1003");
        ...
        Listing fourth = new Listing();
        fourth.setPropertyType("apartment");
        fourth.setId("1001");
        ...
        Map<String, String> meta = new HashMap<>();
        meta.put("contentType", "application/json");

        // 保存状态
        SaveStateRequest request = new SaveStateRequest(STATE_STORE_NAME).setStates(
          new State<>("1", first, null, meta, null),
          new State<>("2", second, null, meta, null),
          new State<>("3", third, null, meta, null),
          new State<>("4", fourth, null, meta, null)
        );
        client.saveBulkState(request).block();
        
        
        // 创建查询和查询状态请求

        Query query = new Query()
          .setFilter(new EqFilter<>("propertyType", "apartment"))
          .setSort(Arrays.asList(new Sorting("id", Sorting.Order.DESC)));
        QueryStateRequest request = new QueryStateRequest(STATE_STORE_NAME)
          .setQuery(query);

        // 使用预览客户端调用查询状态 API
        QueryStateResponse<MyData> result = previewClient.queryState(request, MyData.class).block();
        
        // 查看查询状态响应 
        System.out.println("Found " + result.getResults().size() + " items.");
        for (QueryStateItem<Listing> item : result.getResults()) {
          System.out.println("Key: " + item.getKey());
          System.out.println("Data: " + item.getValue());
        }
}

分布式锁

package io.dapr.examples.lock.grpc;

import io.dapr.client.DaprClientBuilder;
import io.dapr.client.DaprPreviewClient;
import io.dapr.client.domain.LockRequest;
import io.dapr.client.domain.UnlockRequest;
import io.dapr.client.domain.UnlockResponseStatus;
import reactor.core.publisher.Mono;

public class DistributedLockGrpcClient {
  private static final String LOCK_STORE_NAME = "lockstore";

  /**
   * 执行各种方法以检查不同的 API。
   *
   * @param args 参数
   * @throws Exception 抛出异常
   */
  public static void main(String[] args) throws Exception {
    try (DaprPreviewClient client = (new DaprClientBuilder()).buildPreviewClient()) {
      System.out.println("Using preview client...");
      tryLock(client);
      unlock(client);
    }
  }

  /**
   * 尝试获取锁。
   *
   * @param client DaprPreviewClient 对象
   */
  public static void tryLock(DaprPreviewClient client) {
    System.out.println("*******trying to get a free distributed lock********");
    try {
      LockRequest lockRequest = new LockRequest(LOCK_STORE_NAME, "resouce1", "owner1", 5);
      Mono<Boolean> result = client.tryLock(lockRequest);
      System.out.println("Lock result -> " + (Boolean.TRUE.equals(result.block()) ? "SUCCESS" : "FAIL"));
    } catch (Exception ex) {
      System.out.println(ex.getMessage());
    }
  }

  /**
   * 解锁。
   *
   * @param client DaprPreviewClient 对象
   */
  public static void unlock(DaprPreviewClient client) {
    System.out.println("*******unlock a distributed lock********");
    try {
      UnlockRequest unlockRequest = new UnlockRequest(LOCK_STORE_NAME, "resouce1", "owner1");
      Mono<UnlockResponseStatus> result = client.unlock(unlockRequest);
      System.out.println("Unlock result ->" + result.block().name());
    } catch (Exception ex) {
      System.out.println(ex.getMessage());
    }
  }
}

工作流

package io.dapr.examples.workflows;

import io.dapr.workflows.client.DaprWorkflowClient;
import io.dapr.workflows.client.WorkflowState;

import java.time.Duration;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 有关设置说明,请参阅 README。
 */
public class DemoWorkflowClient {

  /**
   * 主方法。
   *
   * @param args 输入参数(未使用)。
   * @throws InterruptedException 如果程序已中断。
   */
  public static void main(String[] args) throws InterruptedException {
    DaprWorkflowClient client = new DaprWorkflowClient();

    try (client) {
      String separatorStr = "*******";
      System.out.println(separatorStr);
      String instanceId = client.scheduleNewWorkflow(DemoWorkflow.class, "input data");
      System.out.printf("Started new workflow instance with random ID: %s%n", instanceId);

      System.out.println(separatorStr);
      System.out.println("**GetWorkflowMetadata:Running Workflow**");
      WorkflowState workflowMetadata = client.getWorkflowState(instanceId, true);
      System.out.printf("Result: %s%n", workflowMetadata);

      System.out.println(separatorStr);
      System.out.println("**WaitForWorkflowStart**");
      try {
        WorkflowState waitForWorkflowStartResult =
            client.waitForWorkflowStart(instanceId, Duration.ofSeconds(60), true);
        System.out.printf("Result: %s%n", waitForWorkflowStartResult);
      } catch (TimeoutException ex) {
        System.out.printf("waitForWorkflowStart has an exception:%s%n", ex);
      }

      System.out.println(separatorStr);
      System.out.println("**SendExternalMessage**");
      client.raiseEvent(instanceId, "TestEvent", "TestEventPayload");

      System.out.println(separatorStr);
      System.out.println("** Registering parallel Events to be captured by allOf(t1,t2,t3) **");
      client.raiseEvent(instanceId, "event1", "TestEvent 1 Payload");
      client.raiseEvent(instanceId, "event2", "TestEvent 2 Payload");
      client.raiseEvent(instanceId, "event3", "TestEvent 3 Payload");
      System.out.printf("Events raised for workflow with instanceId: %s\n", instanceId);

      System.out.println(separatorStr);
      System.out.println("** Registering Event to be captured by anyOf(t1,t2,t3) **");
      client.raiseEvent(instanceId, "e2", "event 2 Payload");
      System.out.printf("Event raised for workflow with instanceId: %s\n", instanceId);


      System.out.println(separatorStr);
      System.out.println("**waitForWorkflowCompletion**");
      try {
        WorkflowState waitForWorkflowCompletionResult =
            client.waitForWorkflowCompletion(instanceId, Duration.ofSeconds(60), true);
        System.out.printf("Result: %s%n", waitForWorkflowCompletionResult);
      } catch (TimeoutException ex) {
        System.out.printf("waitForWorkflowCompletion has an exception:%s%n", ex);
      }

      System.out.println(separatorStr);
      System.out.println("**purgeWorkflow**");
      boolean purgeResult = client.purgeWorkflow(instanceId);
      System.out.printf("purgeResult: %s%n", purgeResult);

      System.out.println(separatorStr);
      System.out.println("**raiseEvent**");

      String eventInstanceId = client.scheduleNewWorkflow(DemoWorkflow.class);
      System.out.printf("Started new workflow instance with random ID: %s%n", eventInstanceId);
      client.raiseEvent(eventInstanceId, "TestException", null);
      System.out.printf("Event raised for workflow with instanceId: %s\n", eventInstanceId);

      System.out.println(separatorStr);
      String instanceToTerminateId = "terminateMe";
      client.scheduleNewWorkflow(DemoWorkflow.class, null, instanceToTerminateId);
      System.out.printf("Started new workflow instance with specified ID: %s%n", instanceToTerminateId);

      TimeUnit.SECONDS.sleep(5);
      System.out.println("Terminate this workflow instance manually before the timeout is reached");
      client.terminateWorkflow(instanceToTerminateId, null);
      System.out.println(separatorStr);

      String restartingInstanceId = "restarting";
      client.scheduleNewWorkflow(DemoWorkflow.class, null, restartingInstanceId);
      System.out.printf("Started new  workflow instance with ID: %s%n", restartingInstanceId);
      System.out.println("Sleeping 30 seconds to restart the workflow");
      TimeUnit.SECONDS.sleep(30);

      System.out.println("**SendExternalMessage: RestartEvent**");
      client.raiseEvent(restartingInstanceId, "RestartEvent", "RestartEventPayload");

      System.out.println("Sleeping 30 seconds to terminate the eternal workflow");
      TimeUnit.SECONDS.sleep(30);
      client.terminateWorkflow(restartingInstanceId, null);
    }

    System.out.println("Exiting DemoWorkflowClient.");
    System.exit(0);
  }
}

边车 API

等待边车

DaprClient 还提供了一个辅助方法来等待边车变得健康(仅限组件)。使用此方法时,请务必指定超时时间(以毫秒为单位)并使用 block() 来等待响应式操作的结果。

// 在尝试使用 Dapr 组件之前,等待 Dapr 边车报告健康。
try (DaprClient client = new DaprClientBuilder().build()) {
  System.out.println("Waiting for Dapr sidecar ...");
  client.waitForSidecar(10000).block(); // 以毫秒为单位指定超时时间
  System.out.println("Dapr sidecar is ready.");
  ...
}

// 在此处执行 Dapr 组件操作,例如获取密钥或保存状态。

关闭边车

try (DaprClient client = new DaprClientBuilder().build()) {
  logger.info("Sending shutdown request.");
  client.shutdown().block();
  logger.info("Ensuring dapr has stopped.");
  ...
}

了解更多有关可添加到 Java 应用程序的 Dapr Java SDK 包的信息。

安全

应用 API 令牌身份验证

像发布订阅、输入绑定或作业这样的构建块需要 Dapr 向您的应用程序发出传入调用,您可以使用 Dapr 应用 API 令牌身份验证来保护这些请求。这确保只有 Dapr 可以调用您应用程序的端点。

了解两种令牌

Dapr 使用两种不同的令牌来保护通信。有关这两种令牌的详细信息,请参阅属性

  • DAPR_API_TOKEN(您的应用 → Dapr 边车):使用 DaprClient 时由 Java SDK 自动处理
  • APP_API_TOKEN(Dapr → 您的应用):需要在应用程序中进行服务器端验证

下面的示例展示如何为 APP_API_TOKEN 实施服务器端验证。

实施服务器端令牌验证

使用 gRPC 协议时,实施服务器拦截器来捕获元数据。

import io.grpc.Context;
import io.grpc.Contexts;
import io.grpc.Metadata;
import io.grpc.ServerCall;
import io.grpc.ServerCallHandler;
import io.grpc.ServerInterceptor;

public class SubscriberGrpcService extends AppCallbackGrpc.AppCallbackImplBase {
  public static final Context.Key<Metadata> METADATA_KEY = Context.key("grpc-metadata");
  
  // gRPC 拦截器来捕获元数据
  public static class MetadataInterceptor implements ServerInterceptor {
    @Override
    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
        ServerCall<ReqT, RespT> call,
        Metadata headers,
        ServerCallHandler<ReqT, RespT> next) {
      Context contextWithMetadata = Context.current().withValue(METADATA_KEY, headers);
      return Contexts.interceptCall(contextWithMetadata, call, headers, next);
    }
  }
  
  // 您的服务方法放在这里...
}

在构建 gRPC 服务器时注册拦截器:

Server server = ServerBuilder.forPort(port)
    .intercept(new SubscriberGrpcService.MetadataInterceptor())
    .addService(new SubscriberGrpcService())
    .build();
server.start();

然后,在您的服务方法中,从元数据中提取令牌:

@Override
public void onTopicEvent(DaprAppCallbackProtos.TopicEventRequest request,
    StreamObserver<DaprAppCallbackProtos.TopicEventResponse> responseObserver) {
  try {
    // 从上下文中提取元数据
    Context context = Context.current();
    Metadata metadata = METADATA_KEY.get(context);
    
    if (metadata != null) {
      String apiToken = metadata.get(
          Metadata.Key.of("dapr-api-token", Metadata.ASCII_STRING_MARSHALLER));
      
      // 相应地验证令牌
    }
    
    // 处理请求
    // ...
    
  } catch (Throwable e) {
    responseObserver.onError(e);
  }
}

与 HTTP 端点一起使用

对于基于 HTTP 的端点,从标头中提取令牌:

@RestController
public class SubscriberController {
  
  @PostMapping(path = "/endpoint")
  public Mono<Void> handleRequest(
      @RequestBody(required = false) byte[] body,
      @RequestHeader Map<String, String> headers) {
    return Mono.fromRunnable(() -> {
      try {
        // 从标头中提取令牌
        String apiToken = headers.get("dapr-api-token");
        
        // 相应地验证令牌
        
        // 处理请求
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    });
  }
}

示例

有关使用发布订阅、绑定和作业的工作示例:

相关链接

有关 SDK 属性的完整列表以及如何配置它们,请访问属性

2.1 - 属性

用于配置 Dapr Java SDK 的全局属性,支持环境变量和系统属性

属性

Dapr Java SDK 提供了一组控制 SDK 行为的全局属性。这些属性可以通过环境变量或系统属性进行配置。系统属性可以在运行 Java 应用程序时使用 -D 标志进行设置。

这些属性会影响整个 SDK,包括客户端和运行时。它们控制的方面包括:

  • 边车连接(端点、端口)
  • 安全设置(TLS、API 令牌)
  • 性能调优(超时、连接池)
  • 协议设置(gRPC、HTTP)
  • 字符串编码

环境变量

以下环境变量可用于配置 Dapr Java SDK:

边车端点

当设置这些变量时,客户端将自动使用它们连接到 Dapr 边车。

环境变量描述默认值
DAPR_GRPC_ENDPOINTDapr 边车的 gRPC 端点localhost:50001
DAPR_HTTP_ENDPOINTDapr 边车的 HTTP 端点localhost:3500
DAPR_GRPC_PORTDapr 边车的 gRPC 端口(已弃用,DAPR_GRPC_ENDPOINT 优先级更高)50001
DAPR_HTTP_PORTDapr 边车的 HTTP 端口(已弃用,DAPR_HTTP_ENDPOINT 优先级更高)3500

API 令牌

Dapr 支持两种类型的 API 令牌来保护通信:

环境变量描述默认值
DAPR_API_TOKEN用于验证从你的应用向 Dapr 边车发起请求的 API 令牌。当使用 DaprClient 时,Java SDK 会自动在请求中包含此令牌。null
APP_API_TOKEN用于验证从 Dapr 向你的应用发起请求的 API 令牌。当设置此变量时,Dapr 在调用你的应用程序时(例如发布订阅订阅者、输入绑定或任务触发器),会在 dapr-api-token 请求头/元数据中包含此令牌。你的应用程序必须验证此令牌。null

有关实现示例,请参阅 App API Token Authentication。更多详情请参阅 Dapr API token authentication

gRPC 配置

TLS 设置

为了安全地进行 gRPC 通信,你可以使用以下环境变量配置 TLS 设置:

环境变量描述默认值
DAPR_GRPC_TLS_INSECURE当设置为 “true” 时,启用不安全的 TLS 模式,该模式仍然使用 TLS 但不验证证书。这会使用 InsecureTrustManagerFactory 来信任所有证书。应仅用于测试或安全环境。false
DAPR_GRPC_TLS_CA_PATHCA 证书文件的路径。用于与具有自签名证书的服务器建立 TLS 连接。null
DAPR_GRPC_TLS_CERT_PATH用于客户端认证的 TLS 证书文件路径。null
DAPR_GRPC_TLS_KEY_PATH用于客户端认证的 TLS 私钥文件路径。null

Keepalive 设置

使用以下环境变量配置 gRPC keepalive 行为:

环境变量描述默认值
DAPR_GRPC_ENABLE_KEEP_ALIVE是否启用 gRPC keepalivefalse
DAPR_GRPC_KEEP_ALIVE_TIME_SECONDSgRPC keepalive 时间(秒)10
DAPR_GRPC_KEEP_ALIVE_TIMEOUT_SECONDSgRPC keepalive 超时(秒)5
DAPR_GRPC_KEEP_ALIVE_WITHOUT_CALLS是否在没有调用时保持 gRPC 连接存活true

入站消息设置

使用以下环境变量配置 gRPC 入站消息设置:

环境变量描述默认值
DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTESDapr 的 gRPC 最大入站消息大小(字节)。此值设置应用程序可以接收的 gRPC 消息的最大大小4194304
DAPR_GRPC_MAX_INBOUND_METADATA_SIZE_BYTESDapr 的 gRPC 最大入站元数据大小(字节)8192

HTTP 客户端配置

这些属性控制用于与 Dapr 边车通信的 HTTP 客户端的行为:

环境变量描述默认值
DAPR_HTTP_CLIENT_READ_TIMEOUT_SECONDSHTTP 客户端读取操作的超时时间(秒)。这是等待 Dapr 边车响应的最长时间。60
DAPR_HTTP_CLIENT_MAX_REQUESTS可以同时执行的最大 HTTP 请求数。超过此限制后,请求将在内存中排队等待正在运行的调用完成。1024
DAPR_HTTP_CLIENT_MAX_IDLE_CONNECTIONSHTTP 连接池中的最大空闲连接数。这是池中可以保持空闲的最大连接数。128

API 配置

这些属性控制通过 SDK 发起的 API 调用的行为:

环境变量描述默认值
DAPR_API_MAX_RETRIES向 Dapr 边车发起 API 调用时,可重试异常的最大重试次数0
DAPR_API_TIMEOUT_MILLISECONDS向 Dapr 边车发起 API 调用的超时时间(毫秒)。值为 0 表示无超时。0

字符串编码

环境变量描述默认值
DAPR_STRING_CHARSETSDK 中用于字符串编码/解码的字符集。必须是有效的 Java 字符集名称。UTF-8

系统属性

所有环境变量都可以使用 -D 标志设置为系统属性。以下是可用的系统属性的完整列表:

系统属性描述默认值
dapr.sidecar.ipDapr 边车的 IP 地址localhost
dapr.http.portDapr 边车的 HTTP 端口3500
dapr.grpc.portDapr 边车的 gRPC 端口50001
dapr.grpc.tls.cert.pathgRPC TLS 证书的路径null
dapr.grpc.tls.key.pathgRPC TLS 密钥的路径null
dapr.grpc.tls.ca.pathgRPC TLS CA 证书的路径null
dapr.grpc.tls.insecure是否使用不安全的 TLS 模式false
dapr.grpc.endpoint远程边车的 gRPC 端点null
dapr.grpc.enable.keep.alive是否启用 gRPC keepalivefalse
dapr.grpc.keep.alive.time.secondsgRPC keepalive 时间(秒)10
dapr.grpc.keep.alive.timeout.secondsgRPC keepalive 超时(秒)5
dapr.grpc.keep.alive.without.calls是否在没有调用时保持 gRPC 连接存活true
dapr.http.endpoint远程边车的 HTTP 端点null
dapr.api.maxRetriesAPI 调用的最大重试次数0
dapr.api.timeoutMillisecondsAPI 调用的超时时间(毫秒)0
dapr.api.token用于身份验证的 API 令牌null
dapr.string.charsetSDK 中使用的字符串编码UTF-8
dapr.http.client.readTimeoutSecondsHTTP 客户端读取的超时时间(秒)60
dapr.http.client.maxRequests最大并发 HTTP 请求数1024
dapr.http.client.maxIdleConnections最大空闲 HTTP 连接数128

属性解析顺序

属性按以下顺序解析:

  1. 覆盖值(如果在创建 Properties 实例时提供)
  2. 系统属性(通过 -D 设置)
  3. 环境变量
  4. 默认值

SDK 按顺序检查每个来源。如果某个值对属性类型无效(例如,数值属性为非数字),SDK 将记录警告并尝试下一个来源。例如:

# 无效的布尔值 - 将被忽略
java -Ddapr.grpc.enable.keep.alive=not-a-boolean -jar myapp.jar

# 有效的布尔值 - 将被使用
export DAPR_GRPC_ENABLE_KEEP_ALIVE=false

在这种情况下,将使用环境变量,因为系统属性值无效。但是,如果两个值都有效,系统属性优先:

# 有效的布尔值 - 将被使用
java -Ddapr.grpc.enable.keep.alive=true -jar myapp.jar

# 有效的布尔值 - 将被忽略
export DAPR_GRPC_ENABLE_KEEP_ALIVE=false

可以通过 DaprClientBuilder 以两种方式设置覆盖值:

  1. 使用单个属性覆盖(大多数情况下推荐):
import io.dapr.config.Properties;

// 设置单个属性覆盖
DaprClient client = new DaprClientBuilder()
    .withPropertyOverride(Properties.GRPC_ENABLE_KEEP_ALIVE, "true")
    .build();

// 或设置多个属性覆盖
DaprClient client = new DaprClientBuilder()
    .withPropertyOverride(Properties.GRPC_ENABLE_KEEP_ALIVE, "true")
    .withPropertyOverride(Properties.HTTP_CLIENT_READ_TIMEOUT_SECONDS, "120")
    .build();
  1. 使用 Properties 实例(当你需要一次设置多个属性时很有用):
// 创建属性覆盖的映射
Map<String, String> overrides = new HashMap<>();
overrides.put("dapr.grpc.enable.keep.alive", "true");
overrides.put("dapr.http.client.readTimeoutSeconds", "120");

// 创建带有覆盖的 Properties 实例
Properties properties = new Properties(overrides);

// 在创建客户端时使用这些属性
DaprClient client = new DaprClientBuilder()
    .withProperties(properties)
    .build();

对于大多数用例,你会使用系统属性或环境变量。覆盖值主要用于在同一应用程序中需要为 SDK 的不同实例设置不同的属性值时。

代理配置

你可以使用系统属性为 Java 应用程序配置代理设置。这些是 Java 网络层(java.net 包)的标准 Java 系统属性,并非 Dapr 特有。它们会被 Java 的网络栈使用,包括 Dapr SDK 使用的 HTTP 客户端。

有关 Java 代理配置的详细信息,包括所有可用属性及其用法,请参阅 Java Networking Properties documentation

例如,以下是配置代理的方法:

# 配置 HTTP 代理 - 替换为你的实际代理服务器详细信息
java -Dhttp.proxyHost=your-proxy-server.com -Dhttp.proxyPort=8080 -jar myapp.jar

# 配置 HTTPS 代理 - 替换为你的实际代理服务器详细信息
java -Dhttps.proxyHost=your-proxy-server.com -Dhttps.proxyPort=8443 -jar myapp.jar

your-proxy-server.com 替换为你的实际代理服务器主机名或 IP 地址,并调整端口号以匹配你的代理服务器配置。

这些代理设置会影响 Java 应用程序发起的所有 HTTP/HTTPS 连接,包括与 Dapr 边车的连接。

3 - 工作流

如何开始使用 Dapr 工作流扩展

3.1 - 操作指南:在 Java SDK 中编写和管理 Dapr 工作流

如何使用 Dapr Java SDK 快速上手工作流

让我们创建一个 Dapr 工作流并通过控制台调用它。借助提供的工作流示例,你将:

本示例使用自托管模式下通过 dapr init 初始化的默认配置。

前置条件

  • 验证你使用的是最新的 proto 绑定

设置环境

克隆 Java SDK 仓库并进入该目录。

git clone https://github.com/dapr/java-sdk.git
cd java-sdk

运行以下命令以安装使用 Dapr Java SDK 运行此工作流示例所需的要求。

mvn clean install

从 Java SDK 根目录进入 Dapr Workflow 示例。

cd examples

运行 DemoWorkflowWorker

DemoWorkflowWorker 类在 Dapr 的工作流运行时引擎中注册 DemoWorkflow 的实现。在 DemoWorkflowWorker.java 文件中,你可以找到 DemoWorkflowWorker 类和 main 方法:

public class DemoWorkflowWorker {

  public static void main(String[] args) throws Exception {
    // 在运行时中注册工作流。
    WorkflowRuntime.getInstance().registerWorkflow(DemoWorkflow.class);
    System.out.println("Start workflow runtime");
    WorkflowRuntime.getInstance().startAndBlock();
    System.exit(0);
  }
}

在上述代码中:

  • WorkflowRuntime.getInstance().registerWorkflow()DemoWorkflow 作为工作流注册到 Dapr Workflow 运行时中。
  • WorkflowRuntime.getInstance().start() 在 Dapr Workflow 运行时内构建并启动引擎。

在终端中,执行以下命令以启动 DemoWorkflowWorker

dapr run --app-id demoworkflowworker --resources-path ./components/workflows --dapr-grpc-port 50001 -- java -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.workflows.DemoWorkflowWorker

预期输出

You're up and running! Both Dapr and your app logs will appear here.

...

== APP == Start workflow runtime
== APP == Sep 13, 2023 9:02:03 AM com.microsoft.durabletask.DurableTaskGrpcWorker startAndBlock
== APP == INFO: Durable Task worker is connecting to sidecar at 127.0.0.1:50001.

运行 DemoWorkflowClient

DemoWorkflowClient 启动已注册到 Dapr 的工作流实例。

public class DemoWorkflowClient {

  // ...
  public static void main(String[] args) throws InterruptedException {
    DaprWorkflowClient client = new DaprWorkflowClient();

    try (client) {
      String separatorStr = "*******";
      System.out.println(separatorStr);
      String instanceId = client.scheduleNewWorkflow(DemoWorkflow.class, "input data");
      System.out.printf("Started new workflow instance with random ID: %s%n", instanceId);

      System.out.println(separatorStr);
      System.out.println("**GetInstanceMetadata:Running Workflow**");
      WorkflowState workflowMetadata = client.getWorkflowState(instanceId, true);
      System.out.printf("Result: %s%n", workflowMetadata);

      System.out.println(separatorStr);
      System.out.println("**WaitForWorkflowStart**");
      try {
        WorkflowState waitForWorkflowStartResult =
            client.waitForWorkflowStart(instanceId, Duration.ofSeconds(60), true);
        System.out.printf("Result: %s%n", waitForWorkflowStartResult);
      } catch (TimeoutException ex) {
        System.out.printf("waitForWorkflowStart has an exception:%s%n", ex);
      }

      System.out.println(separatorStr);
      System.out.println("**SendExternalMessage**");
      client.raiseEvent(instanceId, "TestEvent", "TestEventPayload");

      System.out.println(separatorStr);
      System.out.println("** Registering parallel Events to be captured by allOf(t1,t2,t3) **");
      client.raiseEvent(instanceId, "event1", "TestEvent 1 Payload");
      client.raiseEvent(instanceId, "event2", "TestEvent 2 Payload");
      client.raiseEvent(instanceId, "event3", "TestEvent 3 Payload");
      System.out.printf("Events raised for workflow with instanceId: %s\n", instanceId);

      System.out.println(separatorStr);
      System.out.println("** Registering Event to be captured by anyOf(t1,t2,t3) **");
      client.raiseEvent(instanceId, "e2", "event 2 Payload");
      System.out.printf("Event raised for workflow with instanceId: %s\n", instanceId);


      System.out.println(separatorStr);
      System.out.println("**waitForWorkflowCompletion**");
      try {
        WorkflowState waitForWorkflowCompletionResult =
            client.waitForWorkflowCompletion(instanceId, Duration.ofSeconds(60), true);
        System.out.printf("Result: %s%n", waitForWorkflowCompletionResult);
      } catch (TimeoutException ex) {
        System.out.printf("waitForWorkflowCompletion has an exception:%s%n", ex);
      }

      System.out.println(separatorStr);
      System.out.println("**purgeWorkflow**");
      boolean purgeResult = client.purgeWorkflow(instanceId);
      System.out.printf("purgeResult: %s%n", purgeResult);

      System.out.println(separatorStr);
      System.out.println("**raiseEvent**");

      String eventInstanceId = client.scheduleNewWorkflow(DemoWorkflow.class);
      System.out.printf("Started new workflow instance with random ID: %s%n", eventInstanceId);
      client.raiseEvent(eventInstanceId, "TestException", null);
      System.out.printf("Event raised for workflow with instanceId: %s\n", eventInstanceId);

      System.out.println(separatorStr);
      String instanceToTerminateId = "terminateMe";
      client.scheduleNewWorkflow(DemoWorkflow.class, null, instanceToTerminateId);
      System.out.printf("Started new workflow instance with specified ID: %s%n", instanceToTerminateId);

      TimeUnit.SECONDS.sleep(5);
      System.out.println("Terminate this workflow instance manually before the timeout is reached");
      client.terminateWorkflow(instanceToTerminateId, null);
      System.out.println(separatorStr);

      String restartingInstanceId = "restarting";
      client.scheduleNewWorkflow(DemoWorkflow.class, null, restartingInstanceId);
      System.out.printf("Started new  workflow instance with ID: %s%n", restartingInstanceId);
      System.out.println("Sleeping 30 seconds to restart the workflow");
      TimeUnit.SECONDS.sleep(30);

      System.out.println("**SendExternalMessage: RestartEvent**");
      client.raiseEvent(restartingInstanceId, "RestartEvent", "RestartEventPayload");

      System.out.println("Sleeping 30 seconds to terminate the eternal workflow");
      TimeUnit.SECONDS.sleep(30);
      client.terminateWorkflow(restartingInstanceId, null);
    }

    System.out.println("Exiting DemoWorkflowClient.");
    System.exit(0);
  }
}

在第二个终端窗口中,运行以下命令以启动工作流:

java -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.workflows.DemoWorkflowClient

预期输出

*******
Started new workflow instance with random ID: 0b4cc0d5-413a-4c1c-816a-a71fa24740d4
*******
**GetInstanceMetadata:Running Workflow**
Result: [Name: 'io.dapr.examples.workflows.DemoWorkflow', ID: '0b4cc0d5-413a-4c1c-816a-a71fa24740d4', RuntimeStatus: RUNNING, CreatedAt: 2023-09-13T13:02:30.547Z, LastUpdatedAt: 2023-09-13T13:02:30.699Z, Input: '"input data"', Output: '']
*******
**WaitForWorkflowStart**
Result: [Name: 'io.dapr.examples.workflows.DemoWorkflow', ID: '0b4cc0d5-413a-4c1c-816a-a71fa24740d4', RuntimeStatus: RUNNING, CreatedAt: 2023-09-13T13:02:30.547Z, LastUpdatedAt: 2023-09-13T13:02:30.699Z, Input: '"input data"', Output: '']
*******
**SendExternalMessage**
*******
** Registering parallel Events to be captured by allOf(t1,t2,t3) **
Events raised for workflow with instanceId: 0b4cc0d5-413a-4c1c-816a-a71fa24740d4
*******
** Registering Event to be captured by anyOf(t1,t2,t3) **
Event raised for workflow with instanceId: 0b4cc0d5-413a-4c1c-816a-a71fa24740d4
*******
**WaitForWorkflowCompletion**
Result: [Name: 'io.dapr.examples.workflows.DemoWorkflow', ID: '0b4cc0d5-413a-4c1c-816a-a71fa24740d4', RuntimeStatus: FAILED, CreatedAt: 2023-09-13T13:02:30.547Z, LastUpdatedAt: 2023-09-13T13:02:55.054Z, Input: '"input data"', Output: '']
*******
**purgeWorkflow**
purgeResult: true
*******
**raiseEvent**
Started new workflow instance with random ID: 7707d141-ebd0-4e54-816e-703cb7a52747
Event raised for workflow with instanceId: 7707d141-ebd0-4e54-816e-703cb7a52747
*******
Started new workflow instance with specified ID: terminateMe
Terminate this workflow instance manually before the timeout is reached
*******
Started new  workflow instance with ID: restarting
Sleeping 30 seconds to restart the workflow
**SendExternalMessage: RestartEvent**
Sleeping 30 seconds to terminate the eternal workflow
Exiting DemoWorkflowClient.

发生了什么?

  1. 当你运行 dapr run 时,工作流 worker 将工作流(DemoWorkflow)及其活动注册到 Dapr Workflow 引擎中。
  2. 当你运行 java 时,工作流客户端通过以下活动启动了工作流实例。你可以在运行 dapr run 的终端中查看相应的输出。
    1. 工作流启动,引发三个并行任务,并等待它们完成。
    2. 工作流客户端调用活动,并将 “Hello Activity” 消息发送到控制台。
    3. 工作流超时并被清除。
    4. 工作流客户端使用随机 ID 启动新的工作流实例,使用另一个名为 terminateMe 的工作流实例将其终止,并使用名为 restarting 的工作流重启它。
    5. 然后工作流客户端退出。

后续步骤

高级特性

任务执行键

任务执行键是由 durabletask-java 库生成的唯一标识符。它们存储在 WorkflowActivityContext 中,可用于跟踪和管理工作流活动的执行。它们特别适用于:

  1. 幂等性:确保同一任务的活动不会被执行多次
  2. 状态管理:跟踪活动执行的状态
  3. 错误处理:以受控方式管理重试和失败

以下是在工作流活动中使用任务执行键的示例:

public class TaskExecutionKeyActivity implements WorkflowActivity {
  @Override
  public Object run(WorkflowActivityContext ctx) {
    // 获取此活动的任务执行键
    String taskExecutionKey = ctx.getTaskExecutionKey();
    
    // 使用该键来实现幂等性或状态管理
    // 例如,检查此任务是否已执行
    if (isTaskAlreadyExecuted(taskExecutionKey)) {
      return getPreviousResult(taskExecutionKey);
    }
    
    // 执行活动逻辑
    Object result = executeActivityLogic();
    
    // 使用任务执行键存储结果
    storeResult(taskExecutionKey, result);
    
    return result;
  }
}

4 - 作业

借助 Dapr Jobs 包,您可以从 Java 应用程序与 Dapr Jobs API 交互,以触发未来按照预定计划运行的操作,可选择是否携带有效负载。要开始使用,请浏览 Dapr Jobs 操作指南。

4.1 - 如何:使用 Java SDK 编写和管理 Dapr Jobs

如何使用 Dapr Java SDK 快速上手 Jobs

在本演示中,我们将调度一个 Dapr Job。调度的 Job 将触发同一应用中注册的端点。使用提供的 Jobs 示例,你将:

本示例使用自托管模式dapr init 的默认配置。

前置条件

设置环境

克隆 Java SDK 仓库 并进入该目录。

git clone https://github.com/dapr/java-sdk.git
cd java-sdk

运行以下命令以安装使用 Dapr Java SDK 运行 jobs 示例所需的要求。

mvn clean install -DskipTests

从 Java SDK 根目录进入示例目录。

cd examples

运行 Dapr sidecar。

dapr run --app-id jobsapp --dapr-grpc-port 51439 --dapr-http-port 3500 --app-port 8080

现在,Dapr 正在 http://localhost:3500 监听 HTTP 请求,在 http://localhost:51439 监听内部 Jobs gRPC 请求。

调度和获取 Job

DemoJobsClient 中有调度 Job 的步骤。使用 DaprPreviewClient 调用 scheduleJob 将向 Dapr Runtime 调度一个 Job。

public class DemoJobsClient {

  /**
   * The main method of this app to schedule and get jobs.
   */
  public static void main(String[] args) throws Exception {
    try (DaprPreviewClient client = new DaprClientBuilder().withPropertyOverrides(overrides).buildPreviewClient()) {

      // Schedule a job.
      System.out.println("**** Scheduling a Job with name dapr-jobs-1 *****");
      ScheduleJobRequest scheduleJobRequest = new ScheduleJobRequest("dapr-job-1",
          JobSchedule.fromString("* * * * * *")).setData("Hello World!".getBytes());
      client.scheduleJob(scheduleJobRequest).block();

      System.out.println("**** Scheduling job dapr-jobs-1 completed *****");
    }
  }
}

调用 getJob 以检索先前创建并调度的 Job 详细信息。

client.getJob(new GetJobRequest("dapr-job-1")).block()

使用以下命令运行 DemoJobsClient

java -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.jobs.DemoJobsClient

示例输出

**** Scheduling a Job with name dapr-jobs-1 *****
**** Scheduling job dapr-jobs-1 completed *****
**** Retrieving a Job with name dapr-jobs-1 *****

设置一个在 Job 触发时被调用的端点

DemoJobsSpringApplication 类启动一个 Spring Boot 应用,该应用注册 JobsController 中指定的端点 该端点充当对调度的 Job 请求的回调。

@RestController
public class JobsController {

  /**
   * Handles jobs callback from Dapr.
   *
   * @param jobName name of the job.
   * @param payload data from the job if payload exists.
   * @return Empty Mono.
   */
  @PostMapping("/job/{jobName}")
  public Mono<Void> handleJob(@PathVariable("jobName") String jobName,
                              @RequestBody(required = false) byte[] payload) {
    System.out.println("Job Name: " + jobName);
    System.out.println("Job Payload: " + new String(payload));

    return Mono.empty();
  }
}

参数:

  • jobName:被触发的 Job 的名称。
  • payload:与 Job 关联的可选负载数据(作为字节数组)。

使用以下命令运行 Spring Boot 应用。

java -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.jobs.DemoJobsSpringApplication

示例输出

Job Name: dapr-job-1
Job Payload: Hello World!

删除一个调度的 Job

public class DemoJobsClient {

  /**
   * The main method of this app deletes a job that was previously scheduled.
   */
  public static void main(String[] args) throws Exception {
    try (DaprPreviewClient client = new DaprClientBuilder().buildPreviewClient()) {

      // Delete a job.
      System.out.println("**** Delete a Job with name dapr-jobs-1 *****");
      client.deleteJob(new DeleteJobRequest("dapr-job-1")).block();
    }
  }
}

后续步骤

5 - Dapr 与 Spring Boot 入门

如何开始使用 Dapr 与 Spring Boot

通过结合 Dapr 和 Spring Boot,我们可以创建独立于基础设施的 Java 应用程序,这些应用程序可以部署到不同的环境中,支持广泛的本地和云提供商服务。

首先,我们将从一个涵盖 DaprClientTestcontainers 集成的简单集成开始,然后使用 Spring 和 Spring Boot 机制和编程模型来利用底层的 Dapr API。这有助于团队移除连接到特定环境的基础设施(数据库、键值存储、消息代理、配置/密钥存储等)所需的客户端和驱动程序等依赖项。

将 Dapr 和 Spring Boot 集成添加到您的项目

如果您已经有 Spring Boot 应用程序,可以直接将以下依赖项添加到您的项目中:

	<dependency>
    <groupId>io.dapr.spring</groupId>
		<artifactId>dapr-spring-boot-starter</artifactId>
		<version>1.16.0</version>
	</dependency>
	<dependency>
		<groupId>io.dapr.spring</groupId>
		<artifactId>dapr-spring-boot-starter-test</artifactId>
		<version>1.16.0</version>
		<scope>test</scope>
	</dependency>

您可以在此处找到最新发布的版本

通过添加这些依赖项,您可以:

  • 在应用程序内自动装配 DaprClient 以供使用
  • 使用 Spring Data 和 Messaging 抽象以及编程模型,这些模型在底层使用 Dapr API
  • 通过依赖 Testcontainers 来引导 Dapr 控制平面服务和默认组件,从而改进您的内部开发循环

一旦这些依赖项在您的应用程序中,您就可以依赖 Spring Boot 自动配置来自动装配 DaprClient 实例:

@Autowired
private DaprClient daprClient;

这将连接到默认的 Dapr gRPC 端点 localhost:50001,要求您在应用程序外部启动 Dapr。

您可以在应用程序的任何位置使用 DaprClient 与 Dapr API 交互,例如从 REST 端点内部:

@RestController
public class DemoRestController {
  @Autowired
  private DaprClient daprClient;

  @PostMapping("/store")
  public void storeOrder(@RequestBody Order order){
    daprClient.saveState("kvstore", order.orderId(), order).block();
  }
}

record Order(String orderId, Integer amount){}

如果您希望在 Spring Boot 应用程序外部避免管理 Dapr,可以依赖 Testcontainers 在应用程序旁边引导 Dapr 以用于开发目的。 为此,我们可以创建一个使用 Testcontainers 来引导使用 Dapr API 开发应用程序所需的所有内容的测试配置。

使用 Testcontainers 和 Dapr 集成,我们让 @TestConfiguration 为我们的应用程序引导 Dapr。 请注意,对于此示例,我们正在使用一个名为 kvstore 的 Statestore 组件配置 Dapr,该组件连接到也由 Testcontainers 引导的 PostgreSQL 实例。

@TestConfiguration(proxyBeanMethods = false)
public class DaprTestContainersConfig {
  @Bean
  @ServiceConnection
  public DaprContainer daprContainer(Network daprNetwork, PostgreSQLContainer<?> postgreSQLContainer){
    
    return new DaprContainer("daprio/daprd:1.16.0-rc.5")
            .withAppName("producer-app")
            .withNetwork(daprNetwork)
            .withComponent(new Component("kvstore", "state.postgresql", "v1", STATE_STORE_PROPERTIES))
            .withComponent(new Component("kvbinding", "bindings.postgresql", "v1", BINDING_PROPERTIES))
            .dependsOn(postgreSQLContainer);
  }
}

在测试类路径中,您可以添加一个使用此配置进行测试的新 Spring Boot 应用程序:

@SpringBootApplication
public class TestProducerApplication {

  public static void main(String[] args) {

    SpringApplication
            .from(ProducerApplication::main)
            .with(DaprTestContainersConfig.class)
            .run(args);
  }
  
}

现在您可以使用以下命令启动应用程序:

mvn spring-boot:test-run

运行此命令将启动应用程序,使用提供的测试配置,其中包括 Testcontainers 和 Dapr 集成。在日志中,您应该能够看到为您的应用程序启动了 daprdplacement 服务容器。

除了之前的配置(DaprTestContainersConfig)之外,您的测试不应该测试 Dapr 本身,只测试应用程序暴露的 REST 端点。

利用 Spring 和 Spring Boot 编程模型与 Dapr

Java SDK 允许您与所有 Dapr 构建块 进行接口。 但是,如果您想利用 Spring 和 Spring Boot 编程模型,可以使用 dapr-spring-boot-starter 集成。 这包括 Spring Data(KeyValueTemplateCrudRepository)的实现以及用于生产和消费消息的 DaprMessagingTemplate (类似于 Spring KafkaSpring PulsarSpring AMQP for RabbitMQ)和 Dapr 工作流。

使用 Spring Data CrudRepositoryKeyValueTemplate

您可以使用众所周知的 Spring Data 构造,这些构造依赖于基于 Dapr 的实现。 使用 Dapr,您不需要添加任何与基础设施相关的驱动程序或客户端,使您的 Spring 应用程序更轻量,并且与其运行的环境解耦。

在底层,这些实现使用 Dapr Statestore 和 Binding API。

配置参数

使用 Spring Data 抽象,您可以配置 Dapr 将使用哪些 statestore 和绑定来连接到可用的基础设施。 这可以通过设置以下属性来完成:

dapr.statestore.name=kvstore
dapr.statestore.binding=kvbinding

然后您可以像这样 @Autowire KeyValueTemplateCrudRepository

@RestController
@EnableDaprRepositories
public class OrdersRestController {
  @Autowired
  private OrderRepository repository;
  
  @PostMapping("/orders")
  public void storeOrder(@RequestBody Order order){
    repository.save(order);
  }

  @GetMapping("/orders")
  public Iterable<Order> getAll(){
    return repository.findAll();
  }


}

其中 OrderRepository 在一个扩展 Spring Data CrudRepository 接口的接口中定义:

public interface OrderRepository extends CrudRepository<Order, String> {}

请注意,@EnableDaprRepositories 注释完成了在 CrudRespository 接口下连接 Dapr API 的所有魔术。 因为 Dapr 允许用户从同一个应用程序与不同的 StateStores 交互,作为用户,您需要提供以下 bean 作为 Spring Boot @Configuration

@Configuration
@EnableConfigurationProperties({DaprStateStoreProperties.class})
public class ProducerAppConfiguration {
  
  @Bean
  public KeyValueAdapterResolver keyValueAdapterResolver(DaprClient daprClient, ObjectMapper mapper, DaprStateStoreProperties daprStatestoreProperties) {
    String storeName = daprStatestoreProperties.getName();
    String bindingName = daprStatestoreProperties.getBinding();

    return new DaprKeyValueAdapterResolver(daprClient, mapper, storeName, bindingName);
  }

  @Bean
  public DaprKeyValueTemplate daprKeyValueTemplate(KeyValueAdapterResolver keyValueAdapterResolver) {
    return new DaprKeyValueTemplate(keyValueAdapterResolver);
  }
  
}

使用 Spring Messaging 生产和消费事件

类似于 Spring Kafka、Spring Pulsar 和 Spring AMQP,您可以使用 DaprMessagingTemplate 将消息发布到配置的基础设施。要消费消息,您可以使用 @Topic 注释(很快将重命名为 @DaprListener)。

要发布事件/消息,您可以在 Spring 应用程序中 @Autowired DaprMessagingTemplate。 对于此示例,我们将发布 Order 事件,并将消息发送到名为 topic 的主题。

@Autowired
private DaprMessagingTemplate<Order> messagingTemplate;

@PostMapping("/orders")
public void storeOrder(@RequestBody Order order){
  repository.save(order);
  messagingTemplate.send("topic", order);
}

CrudRepository 类似,我们需要指定要使用哪个 PubSub 代理来发布和消费我们的消息。

dapr.pubsub.name=pubsub

因为使用 Dapr 您可以连接到多个 PubSub 代理,您需要提供以下 bean 以让 Dapr 知道您的 DaprMessagingTemplate 将使用哪个 PubSub 代理:

@Bean
public DaprMessagingTemplate<Order> messagingTemplate(DaprClient daprClient,
                                                             DaprPubSubProperties daprPubSubProperties) {
  return new DaprMessagingTemplate<>(daprClient, daprPubSubProperties.getName());
}

最后,因为 Dapr PubSub 需要在您的应用程序和 Dapr 之间建立双向连接,所以您需要使用几个参数扩展您的 Testcontainers 配置:

@Bean
@ServiceConnection
public DaprContainer daprContainer(Network daprNetwork, PostgreSQLContainer<?> postgreSQLContainer, RabbitMQContainer rabbitMQContainer){
    
    return new DaprContainer("daprio/daprd:1.16.0-rc.5")
            .withAppName("producer-app")
            .withNetwork(daprNetwork)
            .withComponent(new Component("kvstore", "state.postgresql", "v1", STATE_STORE_PROPERTIES))
            .withComponent(new Component("kvbinding", "bindings.postgresql", "v1", BINDING_PROPERTIES))
            .withComponent(new Component("pubsub", "pubsub.rabbitmq", "v1", rabbitMqProperties))
            .withAppPort(8080)
            .withAppChannelAddress("host.testcontainers.internal")
            .dependsOn(rabbitMQContainer)
            .dependsOn(postgreSQLContainer);
}

现在,在 Dapr 配置中,我们包含了一个 pubsub 组件,它将连接到由 Testcontainers 启动的 RabbitMQ 实例。 我们还设置了两个重要参数 .withAppPort(8080).withAppChannelAddress("host.testcontainers.internal"),这允许 Dapr 在代理中发布消息时联系回应用程序。

要监听事件/消息,您需要在应用程序中暴露一个负责接收消息的端点。 如果您暴露 REST 端点,可以使用 @Topic 注释让 Dapr 知道它也需要将事件/消息转发到哪里:

@PostMapping("subscribe")
@Topic(pubsubName = "pubsub", name = "topic")
public void subscribe(@RequestBody CloudEvent<Order> cloudEvent){
    events.add(cloudEvent);
}

在引导应用程序时,Dapr 将注册要转发到您的应用程序暴露的 subscribe 端点的消息订阅。

如果您正在为这些订阅者编写测试,您需要确保 Testcontainers 知道您的应用程序将在端口 8080 上运行, 因此使用 Testcontainers 启动的容器知道您的应用程序在哪里:

@BeforeAll
public static void setup(){
  org.testcontainers.Testcontainers.exposeHostPorts(8080);
}

您可以在此处查看并运行完整的示例源代码

后续步骤

了解有关可添加到您的 Java 应用程序的 Dapr Java SDK 包的更多信息。

查看如何指南,使用 Spring Boot 和 Testcontainers 进行 Dapr 工作流以获得本地工作流开发体验

相关链接

5.1 - 操作指南:使用 Spring Boot 编写和管理 Dapr 工作流

如何使用 Spring Boot 集成快速上手工作流

遵循与 Spring Data 和 Spring Messaging 相同的方法,dapr-spring-boot-starter 为 Spring Boot 用户带来了 Dapr 工作流集成。

使用 Dapr 工作流,您可以在 Java 代码中定义复杂的编排(工作流)。Dapr Spring Boot Starter 通过将 Workflows 和 WorkflowActivitys 作为 Spring Bean 进行管理,使您的开发更加便捷。

为了启用自动 bean 发现,您需要在 @SpringBootApplication 上添加 @EnableDaprWorkflows 注解:

@SpringBootApplication
@EnableDaprWorkflows
public class MySpringBootApplication {
  ...
}

通过添加此注解,所有的 Workflows 和 WorkflowActivitys bean 都会被 Spring 自动发现并注册到工作流引擎中。

创建工作流和活动

在您的 Spring Boot 应用程序中,您可以定义任意数量的工作流。为此,您需要创建新的 Workflow 接口实现。

@Component
public class MyWorkflow implements Workflow {
    
    @Override
    public WorkflowStub create() {
      return ctx -> {
         <工作流逻辑>
      };
    }

}

在工作流定义内部,您可以执行服务间交互、调度定时器或接收外部事件。

通过将所有 WorkflowActivitys 作为托管 bean,您可以使用 Spring 的 @Autowired 机制来注入工作流活动实现其功能所需的任何 bean。例如 @RestTemplate

@Component
public class MyWorkflowActivity implements WorkflowActivity {

  @Autowired
  private RestTemplate restTemplate;

创建和与工作流交互

要创建和与工作流实例交互,您可以使用同样支持 @AutowiredDaprWorkflowClient

@Autowired
private DaprWorkflowClient daprWorkflowClient;

应用程序现在可以调度新的工作流实例并触发事件。

String instanceId = daprWorkflowClient.scheduleNewWorkflow(MyWorkflow.class, payload);

以及

daprWorkflowClient.raiseEvent(instanceId, "MyEvenet", event);

在此处查看完整示例

后续步骤和资源

查看 Baeldung 关于 Dapr 工作流和 Dapr 发布订阅的博客文章,其中包含完整的工作示例。

查看 Dapr 工作流文档,了解如何使用 Dapr 工作流的更多信息。