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 根目录进入示例目录。
运行 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 应用程序进行交互。
注意
如果您还没有,请
尝试其中一个快速入门,快速了解如何将 Dapr Java SDK 与 API 构建块一起使用。
前置条件
完成初始设置并将 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_ENDPOINT | Dapr 边车的 gRPC 端点 | localhost:50001 |
DAPR_HTTP_ENDPOINT | Dapr 边车的 HTTP 端点 | localhost:3500 |
DAPR_GRPC_PORT | Dapr 边车的 gRPC 端口(已弃用,DAPR_GRPC_ENDPOINT 优先级更高) | 50001 |
DAPR_HTTP_PORT | Dapr 边车的 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_PATH | CA 证书文件的路径。用于与具有自签名证书的服务器建立 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 keepalive | false |
DAPR_GRPC_KEEP_ALIVE_TIME_SECONDS | gRPC keepalive 时间(秒) | 10 |
DAPR_GRPC_KEEP_ALIVE_TIMEOUT_SECONDS | gRPC keepalive 超时(秒) | 5 |
DAPR_GRPC_KEEP_ALIVE_WITHOUT_CALLS | 是否在没有调用时保持 gRPC 连接存活 | true |
入站消息设置
使用以下环境变量配置 gRPC 入站消息设置:
| 环境变量 | 描述 | 默认值 |
|---|
DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES | Dapr 的 gRPC 最大入站消息大小(字节)。此值设置应用程序可以接收的 gRPC 消息的最大大小 | 4194304 |
DAPR_GRPC_MAX_INBOUND_METADATA_SIZE_BYTES | Dapr 的 gRPC 最大入站元数据大小(字节) | 8192 |
HTTP 客户端配置
这些属性控制用于与 Dapr 边车通信的 HTTP 客户端的行为:
| 环境变量 | 描述 | 默认值 |
|---|
DAPR_HTTP_CLIENT_READ_TIMEOUT_SECONDS | HTTP 客户端读取操作的超时时间(秒)。这是等待 Dapr 边车响应的最长时间。 | 60 |
DAPR_HTTP_CLIENT_MAX_REQUESTS | 可以同时执行的最大 HTTP 请求数。超过此限制后,请求将在内存中排队等待正在运行的调用完成。 | 1024 |
DAPR_HTTP_CLIENT_MAX_IDLE_CONNECTIONS | HTTP 连接池中的最大空闲连接数。这是池中可以保持空闲的最大连接数。 | 128 |
API 配置
这些属性控制通过 SDK 发起的 API 调用的行为:
| 环境变量 | 描述 | 默认值 |
|---|
DAPR_API_MAX_RETRIES | 向 Dapr 边车发起 API 调用时,可重试异常的最大重试次数 | 0 |
DAPR_API_TIMEOUT_MILLISECONDS | 向 Dapr 边车发起 API 调用的超时时间(毫秒)。值为 0 表示无超时。 | 0 |
字符串编码
| 环境变量 | 描述 | 默认值 |
|---|
DAPR_STRING_CHARSET | SDK 中用于字符串编码/解码的字符集。必须是有效的 Java 字符集名称。 | UTF-8 |
系统属性
所有环境变量都可以使用 -D 标志设置为系统属性。以下是可用的系统属性的完整列表:
| 系统属性 | 描述 | 默认值 |
|---|
dapr.sidecar.ip | Dapr 边车的 IP 地址 | localhost |
dapr.http.port | Dapr 边车的 HTTP 端口 | 3500 |
dapr.grpc.port | Dapr 边车的 gRPC 端口 | 50001 |
dapr.grpc.tls.cert.path | gRPC TLS 证书的路径 | null |
dapr.grpc.tls.key.path | gRPC TLS 密钥的路径 | null |
dapr.grpc.tls.ca.path | gRPC TLS CA 证书的路径 | null |
dapr.grpc.tls.insecure | 是否使用不安全的 TLS 模式 | false |
dapr.grpc.endpoint | 远程边车的 gRPC 端点 | null |
dapr.grpc.enable.keep.alive | 是否启用 gRPC keepalive | false |
dapr.grpc.keep.alive.time.seconds | gRPC keepalive 时间(秒) | 10 |
dapr.grpc.keep.alive.timeout.seconds | gRPC keepalive 超时(秒) | 5 |
dapr.grpc.keep.alive.without.calls | 是否在没有调用时保持 gRPC 连接存活 | true |
dapr.http.endpoint | 远程边车的 HTTP 端点 | null |
dapr.api.maxRetries | API 调用的最大重试次数 | 0 |
dapr.api.timeoutMilliseconds | API 调用的超时时间(毫秒) | 0 |
dapr.api.token | 用于身份验证的 API 令牌 | null |
dapr.string.charset | SDK 中使用的字符串编码 | UTF-8 |
dapr.http.client.readTimeoutSeconds | HTTP 客户端读取的超时时间(秒) | 60 |
dapr.http.client.maxRequests | 最大并发 HTTP 请求数 | 1024 |
dapr.http.client.maxIdleConnections | 最大空闲 HTTP 连接数 | 128 |
属性解析顺序
属性按以下顺序解析:
- 覆盖值(如果在创建 Properties 实例时提供)
- 系统属性(通过
-D 设置) - 环境变量
- 默认值
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 以两种方式设置覆盖值:
- 使用单个属性覆盖(大多数情况下推荐):
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();
- 使用 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.1 - 操作指南:在 Java SDK 中编写和管理 Dapr 工作流
如何使用 Dapr Java SDK 快速上手工作流
让我们创建一个 Dapr 工作流并通过控制台调用它。借助提供的工作流示例,你将:
本示例使用自托管模式下通过 dapr init 初始化的默认配置。
前置条件
设置环境
克隆 Java SDK 仓库并进入该目录。
git clone https://github.com/dapr/java-sdk.git
cd java-sdk
运行以下命令以安装使用 Dapr Java SDK 运行此工作流示例所需的要求。
从 Java SDK 根目录进入 Dapr Workflow 示例。
运行 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.
发生了什么?
- 当你运行
dapr run 时,工作流 worker 将工作流(DemoWorkflow)及其活动注册到 Dapr Workflow 引擎中。 - 当你运行
java 时,工作流客户端通过以下活动启动了工作流实例。你可以在运行 dapr run 的终端中查看相应的输出。- 工作流启动,引发三个并行任务,并等待它们完成。
- 工作流客户端调用活动,并将 “Hello Activity” 消息发送到控制台。
- 工作流超时并被清除。
- 工作流客户端使用随机 ID 启动新的工作流实例,使用另一个名为
terminateMe 的工作流实例将其终止,并使用名为 restarting 的工作流重启它。 - 然后工作流客户端退出。
后续步骤
高级特性
任务执行键
任务执行键是由 durabletask-java 库生成的唯一标识符。它们存储在 WorkflowActivityContext 中,可用于跟踪和管理工作流活动的执行。它们特别适用于:
- 幂等性:确保同一任务的活动不会被执行多次
- 状态管理:跟踪活动执行的状态
- 错误处理:以受控方式管理重试和失败
以下是在工作流活动中使用任务执行键的示例:
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.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 根目录进入示例目录。
运行 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 应用程序,这些应用程序可以部署到不同的环境中,支持广泛的本地和云提供商服务。
首先,我们将从一个涵盖 DaprClient 和 Testcontainers 集成的简单集成开始,然后使用 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 和 DaprWorkflowClient 预配置的:
dapr.client.httpEndpoint=http://localhost
dapr.client.httpPort=3500
dapr.client.grpcEndpoint=localhost
dapr.client.grpcPort=50001
dapr.client.apiToken=<your remote api token>
这些值是默认使用的,但您可以在 application.properties 文件中覆盖它们以适应您的环境。请注意,同时支持 kebab-case 和 camelCase。
您可以在应用程序的任何位置使用 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);
}
}
现在您可以使用以下命令启动应用程序:
运行此命令将启动应用程序,使用提供的测试配置,其中包括 Testcontainers 和 Dapr 集成。在日志中,您应该能够看到为您的应用程序启动了 daprd 和 placement 服务容器。
除了之前的配置(DaprTestContainersConfig)之外,您的测试不应该测试 Dapr 本身,只测试应用程序暴露的 REST 端点。
利用 Spring 和 Spring Boot 编程模型与 Dapr
Java SDK 允许您与所有 Dapr 构建块 进行接口。
但是,如果您想利用 Spring 和 Spring Boot 编程模型,可以使用 dapr-spring-boot-starter 集成。
这包括 Spring Data(KeyValueTemplate 和 CrudRepository)的实现以及用于生产和消费消息的 DaprMessagingTemplate
(类似于 Spring Kafka、Spring Pulsar 和 Spring AMQP for RabbitMQ)和 Dapr 工作流。
使用 Spring Data CrudRepository 和 KeyValueTemplate
您可以使用众所周知的 Spring Data 构造,这些构造依赖于基于 Dapr 的实现。
使用 Dapr,您不需要添加任何与基础设施相关的驱动程序或客户端,使您的 Spring 应用程序更轻量,并且与其运行的环境解耦。
在底层,这些实现使用 Dapr Statestore 和 Binding API。
配置参数
使用 Spring Data 抽象,您可以配置 Dapr 将使用哪些 statestore 和绑定来连接到可用的基础设施。
这可以通过设置以下属性来完成:
dapr.statestore.name=kvstore
dapr.statestore.binding=kvbinding
然后您可以像这样 @Autowire KeyValueTemplate 或 CrudRepository:
@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 代理,您需要提供以下 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;
创建和与工作流交互
要创建和与工作流实例交互,您可以使用同样支持 @Autowired 的 DaprWorkflowClient。
@Autowired
private DaprWorkflowClient daprWorkflowClient;
应用程序现在可以调度新的工作流实例并触发事件。
String instanceId = daprWorkflowClient.scheduleNewWorkflow(MyWorkflow.class, payload);
以及
daprWorkflowClient.raiseEvent(instanceId, "MyEvenet", event);
在此处查看完整示例
后续步骤和资源
查看 Baeldung 关于 Dapr 工作流和 Dapr 发布订阅的博客文章,其中包含完整的工作示例。
查看 Dapr 工作流文档,了解如何使用 Dapr 工作流的更多信息。