Camel REST：长期运行的 HTTP 服务的异步请求处理
Camel REST: Async request processing for long-operating HTTP services
我试图找到此类问题的最新示例，但不幸的是没有找到。
我正在尝试使用 camel 实现一个 web 服务，它的行为应该如下所示：
- Camel 通过 GET 或 POST (api/startsearch)
从 Rest-Endpoint 接收输入
- 一个 bean 处理输入并生成一个 ticket-id
- 同一个 bean 使用 HTTP-202 或重定向状态代码响应客户端，包括重定向 url (api/result?ticket-id=jf3298u23)。
- bean 还将输入传递给 activemq 启动队列，其中 Camel 路由将完成其所有长操作处理。
- 路由完成后，结果应该在重定向 URL (/result?ticket-id=jf3298u23) 处可用。如果处理尚未完成，它应该使用自定义状态代码进行响应，例如 HTTP-299-processing。
所以我的路线是这样的：
rest().path(apiPath).produces("application/json")
.get(searchEndpoint)
.to("bean:requestHandler?method=processNewSearch") // generate ticket-id and reply with 202 or 3xx
.route().inOnly("activemq:queue:start").endRest() // put the incoming message into the start-queue where the processing starts
.get(resultEndpoint).to("bean:requestHandler?method=returnResult"); // return 299 when processing not done or 200 + result
from("activemq:queue:start")
.setHeader("recipients").method(new ExtractRecipients(), "extractRecipients")
.to("activemq:queue:recipientlist");
... etc, etc... until:
from("activemq:queue:output")
.to("bean:requestHandler?method=saveFinishedSearch");
bean本身有3个方法：
public void processNewSearch(Exchange exchange) {
//generate ticket and stuff and finally set Header and body of the response
exchange.getOut().setHeader(Exchange.HTTP_RESPONSE_CODE, 202);
exchange.getOut().setBody(redirectUrl);
}
public void returnResult(Exchange exchange) {
//handle ticket related stuff, if valid fetch result and create http response:
exchange.getOut().setHeader(Exchange.HTTP_RESPONSE_CODE, 200);
exchange.getOut().setBody(searchResult);
return;
}
public void saveFinishedSearch(Exchange exchange) {
// get search Results from the final Message that was processing asynchronously in the background and save it
finishedSearches.put(ticket, body);
}
我确定这不是使用手动设置的响应代码和消息进行回复的正确方法，但我没有找到其他方法。
所以目前的问题是 camel 一直等到整个消息被处理，因此 .to("bean:requestHandler?method=processNewSearch") 生成的响应什么都不做，因为它只会被放入启动队列。
如何使用 camel 立即 return 自定义响应并让路由异步处理请求？
首先，您应该坚持 HTTP Protocol 并且仅通过 POST 操作触发后台任务。您可能不希望爬虫通过 GET 请求触发较长的 运行 后台进程，是吗？
因此，您还应该使用 Location HTTP header 到 return 资源的 URI 有关进程当前状态的更多信息可以从.我也会使用通用 URI 而不是某些重定向。
在你的 route-setup 中，我通常也会将所有路由相关的东西都放在 .route() 块中。我们维护一个目录组装过程和一个 EDI 消息存档系统，由于德国法律强制客户备份他们交换的 EDI 消息，该系统将 assemble 在特定时间范围内发送 and/or 接收的消息。
我们将触发新的存档或组装请求与检索请求的当前状态分开。
rest("/archives")
.post()
.bindingMode(RestBindingMode.json)
.type(ArchiveRequestSettings.class)
.consumes(MediaType.APPLICATION_JSON)
.produces(MediaType.APPLICATION_JSON)
.description("Invokes the generation of a new message archive for
"messages matching a criteria contained in the payload")
.route().routeId("create-archives")
// Extract the IP address of the user who invokes the service
.bean(ExtractClientIP.class)
// Basic Authentication
.bean(SpringSecurityContextLoader.class).policy(authorizationPolicy)
// check the amount of requests received within a certain time-period
.bean(receivedRequestFilter)
// extract specified settings
.bean(ExtractArchiveRequestSettings.class)
// forward the task to the archive generation queue
.to(SomeEndpoints.ARCHIVE_GENERATION_QUEUE)
// return 202 Accepted response
.bean(ReturnArchiveRequestCreatedStatus.class)
.endRest()
.get("/{archiveId}")
.bindingMode(RestBindingMode.json)
.outType(ArchiveRequestEntity.class)
.produces(MediaType.APPLICATION_JSON)
.description("Returns the status of the message archive generation process."
+ " If the process has finished this operation will return the"
+ " link to the download location of the generated archive")
.route().routeId("archive-status")
// Extract the IP address of the user who invokes the service
.bean(ExtractClientIP.class)
// Basic Authentication
.bean(SpringSecurityContextLoader.class).policy(authorizationPolicy)
// check the amount of requests received within a certain time-period
.bean(receivedRequestFilter)
// return the current state of the task to the client. If the job is done,
// the response will also include a download link as wel as an MD5 hash to
// verify the correctness of the downloaded archive
.bean(ReturnArchiveRequestStatus.class)
.endRest();
ExtractArchiveRequestSettings class 仅对接收到的有效负载执行健全性检查并为缺失字段设置默认值。之后，请求被存储到数据库中，其唯一标识符存储到 header.
中
ArchiveRequestSetting 看起来确实像下面的示例（稍微简化）
@Getter
@Setter
@JsonIgnoreProperties(ignoreUnknown = true)
@JsonInclude(JsonInclude.Include.NON_NULL)
public class ArchiveRequestSettings {
/** Specifies if sent or received messages should be included in the artifact. Setting this field
* to 'DELIVERED' will include only delivered documents where the companyUuid of the requesting
* user matches the documents sender identifier. Specifying this field as 'RECEIVED' will include
* only documents whose receiver identifier matches the companyUuid of the requesting user. **/
private String direction;
/** The naming schema of entries within the archive **/
private String entryPattern;
/** The upper timestamp bound to include messages. Entries older than this value will be omitted **/
@JsonSerialize(using = Iso8601DateSerializer.class)
@JsonDeserialize(using = Iso8601DateDeserializer.class)
private Date from;
/** The lower timestamp bound to include messages. Entries younger than this value will be
* omitted. If left empty this will include even the most recent messages. **/
@JsonSerialize(using = Iso8601DateSerializer.class)
@JsonDeserialize(using = Iso8601DateDeserializer.class)
private Date till;
}
ReturnArchiveRequestCreatedStatus class 查找存储的请求实体并 return 使用 202 Accepted 响应对其进行处理。
@Handler
public void returnStatus(Exchange exchange) {
String archiveId = exchange.getIn().getHeader(HeaderConstants.ARCHIVES_REQUEST_ID, String.class);
ArchiveRequestEntity archive = repository.findOne(archiveId);
Message msg = new DefaultMessage(exchange.getContext());
msg.setHeader(Exchange.HTTP_RESPONSE_CODE, 202); // Accepted
msg.setHeader(Exchange.CONTENT_TYPE, "application/json; charset=\"utf-8\"");
msg.setHeader("Location", archiveLocationUrl + "/" + archiveId);
msg.setBody(archive);
exchange.setOut(msg);
}
返回存储请求的当前状态可确保客户端可以检查实际应用了哪些设置，并可以在某些默认设置不方便或需要应用进一步更改时更新它们。
实际的支持过程是通过将交换发送到 Redis queue 开始的，该 Redis queue 在不同的机器上使用。此过程的输出将是一个包含请求文件的存档，该文件将上传到 public 可访问位置，并且只有 link 将存储在请求实体中。请注意，我们有一个自定义的 camel 组件，它模仿了 Redis queues 的 seda entpoint。使用 seda 虽然应该足以在不同的线程中开始处理任务。
根据支持进程的当前状态，存储的请求实体将由支持进程更新。在收到状态请求（通过 GET）时，查询数据存储的当前状态并映射到某些响应：
public class ReturnArchiveRequestStatus {
@Resource
private ArchiveRequestRepository repository;
@Handler
public void returnArchiveStatus(Exchange exchange) throws JSONException {
String archiveId = exchange.getIn().getHeader("archiveId", String.class);
if (StringUtils.isBlank(archiveId)) {
badRequest(exchange);
return;
}
ArchiveRequestEntity archive = repository.findOne(archiveId);
if (null == archive) {
notFound(archiveId, exchange);
return;
}
ok(archive, exchange);
}
private void badRequest(Exchange exchange) throws JSONException {
Message msg = new DefaultMessage(exchange.getContext());
msg.setHeader(Exchange.HTTP_RESPONSE_CODE, 400);
msg.setHeader(Exchange.CONTENT_TYPE, "application/json; charset=\"utf-8\"");
msg.setFault(false);
JSONObject json = new JSONObject();
json.put("status", "ERROR");
json.put("message", "No archive identifier found");
msg.setBody(json.toString());
exchange.setOut(msg);
}
private void notFound(String archiveId, Exchange exchange) throws JSONException {
Message msg = new DefaultMessage(exchange.getContext());
msg.setHeader(Exchange.HTTP_RESPONSE_CODE, 403);
msg.setHeader(Exchange.CONTENT_TYPE, "application/json; charset=\"utf-8\"");
msg.setFault(false);
JSONObject json = new JSONObject();
json.put("status", "ERROR");
json.put("message", "Could not find pending archive process with ID " + archiveId);
msg.setBody(json.toString());
exchange.setOut(msg);
}
private void ok(UserArchiveRequestEntity archive, Exchange exchange) throws JSONException {
Message msg = new DefaultMessage(exchange.getContext());
msg.setHeader(Exchange.HTTP_RESPONSE_CODE, 200);
msg.setHeader(Exchange.CONTENT_TYPE, "application/json; charset=\"utf-8\"");
msg.setFault(false);
msg.setBody(archive);
exchange.setOut(msg);
}
}
在整个过程中存储和更新的实际实体看起来是这样的（简化）：
@Getter
@Setter
@Builder
@ToString
@Document(collection = "archive")
@JsonInclude(JsonInclude.Include.NON_EMPTY)
public class ArchiveRequestEntity {
/**
* The current state of the archiving process
*/
public enum State {
/** The request to create an archive was cued but not yet processed **/
QUEUED,
/** The archive is currently under construction **/
RUNNING,
/** The archive was generated successfully. {@link #downloadUrl} should contain the link the
* archive can be found **/
FINISHED,
/** Indicates that the archive generation failed. {@link #error} should indicate the actual
* reason why the request failed **/
FAILED
}
@Id
@JsonIgnore
private String id;
/** Timestamp the process was triggered **/
@JsonIgnore
@Indexed(expireAfterSeconds = DEFAULT_EXPIRE_TIME)
private Date timestamp = new Date();
/** The identifier of the company to create the archive for **/
private String companyUuid;
/** The state this archive is currently in **/
private State state = State.QUEUED;
...
/** Marks the upper limit to include entries to the archive. Entries older then this field will
* not be included in the archives while entries equal or younger than this timestamp will be
* included unless they are younger than {@link #till} timestamp **/
@JsonFormat(pattern = "yyyy-MM-dd'T'HH:mm:ssXX")
private Date from;
/** Marks the lower limit to include entries to the archive. Entries younger than this field will
* not be included in the archive **/
@JsonFormat(pattern = "yyyy-MM-dd'T'HH:mm:ssXX")
private Date till;
/** Information on why the archive creation failed **/
private String error;
/** The URL of the final archive to download **/
private String downloadUrl;
/** The MD5 Hash of the final artifact in order to guarantee clients an unmodified version of the
* archive **/
private String md5Hash;
...
}
请注意，无论处理状态的当前状态如何，200 OK 都将 return 与当前的 JSON 处理状态表示形式相结合。客户端将看到具有 downloadUrl 和 md5Hash 属性集的 FINISHED 状态，或者看到具有不同属性的不同状态。
当然，支持进程需要适当地更新请求状态，否则客户端将无法检索到有关请求当前状态的正确信息。
这种方法应该适用于几乎所有长 运行 进程，尽管您传递的信息的内部结构可能与我们的场景不同。希望这有帮助
我试图找到此类问题的最新示例，但不幸的是没有找到。 我正在尝试使用 camel 实现一个 web 服务，它的行为应该如下所示：
- Camel 通过 GET 或 POST (api/startsearch) 从 Rest-Endpoint 接收输入
- 一个 bean 处理输入并生成一个 ticket-id
- 同一个 bean 使用 HTTP-202 或重定向状态代码响应客户端，包括重定向 url (api/result?ticket-id=jf3298u23)。
- bean 还将输入传递给 activemq 启动队列，其中 Camel 路由将完成其所有长操作处理。
- 路由完成后，结果应该在重定向 URL (/result?ticket-id=jf3298u23) 处可用。如果处理尚未完成，它应该使用自定义状态代码进行响应，例如 HTTP-299-processing。
所以我的路线是这样的：
rest().path(apiPath).produces("application/json")
.get(searchEndpoint)
.to("bean:requestHandler?method=processNewSearch") // generate ticket-id and reply with 202 or 3xx
.route().inOnly("activemq:queue:start").endRest() // put the incoming message into the start-queue where the processing starts
.get(resultEndpoint).to("bean:requestHandler?method=returnResult"); // return 299 when processing not done or 200 + result
from("activemq:queue:start")
.setHeader("recipients").method(new ExtractRecipients(), "extractRecipients")
.to("activemq:queue:recipientlist");
... etc, etc... until:
from("activemq:queue:output")
.to("bean:requestHandler?method=saveFinishedSearch");
bean本身有3个方法：
public void processNewSearch(Exchange exchange) {
//generate ticket and stuff and finally set Header and body of the response
exchange.getOut().setHeader(Exchange.HTTP_RESPONSE_CODE, 202);
exchange.getOut().setBody(redirectUrl);
}
public void returnResult(Exchange exchange) {
//handle ticket related stuff, if valid fetch result and create http response:
exchange.getOut().setHeader(Exchange.HTTP_RESPONSE_CODE, 200);
exchange.getOut().setBody(searchResult);
return;
}
public void saveFinishedSearch(Exchange exchange) {
// get search Results from the final Message that was processing asynchronously in the background and save it
finishedSearches.put(ticket, body);
}
我确定这不是使用手动设置的响应代码和消息进行回复的正确方法，但我没有找到其他方法。
所以目前的问题是 camel 一直等到整个消息被处理，因此 .to("bean:requestHandler?method=processNewSearch") 生成的响应什么都不做，因为它只会被放入启动队列。
如何使用 camel 立即 return 自定义响应并让路由异步处理请求？
首先，您应该坚持 HTTP Protocol 并且仅通过 POST 操作触发后台任务。您可能不希望爬虫通过 GET 请求触发较长的 运行 后台进程，是吗？
因此，您还应该使用 Location HTTP header 到 return 资源的 URI 有关进程当前状态的更多信息可以从.我也会使用通用 URI 而不是某些重定向。
在你的 route-setup 中，我通常也会将所有路由相关的东西都放在 .route() 块中。我们维护一个目录组装过程和一个 EDI 消息存档系统，由于德国法律强制客户备份他们交换的 EDI 消息，该系统将 assemble 在特定时间范围内发送 and/or 接收的消息。
我们将触发新的存档或组装请求与检索请求的当前状态分开。
rest("/archives")
.post()
.bindingMode(RestBindingMode.json)
.type(ArchiveRequestSettings.class)
.consumes(MediaType.APPLICATION_JSON)
.produces(MediaType.APPLICATION_JSON)
.description("Invokes the generation of a new message archive for
"messages matching a criteria contained in the payload")
.route().routeId("create-archives")
// Extract the IP address of the user who invokes the service
.bean(ExtractClientIP.class)
// Basic Authentication
.bean(SpringSecurityContextLoader.class).policy(authorizationPolicy)
// check the amount of requests received within a certain time-period
.bean(receivedRequestFilter)
// extract specified settings
.bean(ExtractArchiveRequestSettings.class)
// forward the task to the archive generation queue
.to(SomeEndpoints.ARCHIVE_GENERATION_QUEUE)
// return 202 Accepted response
.bean(ReturnArchiveRequestCreatedStatus.class)
.endRest()
.get("/{archiveId}")
.bindingMode(RestBindingMode.json)
.outType(ArchiveRequestEntity.class)
.produces(MediaType.APPLICATION_JSON)
.description("Returns the status of the message archive generation process."
+ " If the process has finished this operation will return the"
+ " link to the download location of the generated archive")
.route().routeId("archive-status")
// Extract the IP address of the user who invokes the service
.bean(ExtractClientIP.class)
// Basic Authentication
.bean(SpringSecurityContextLoader.class).policy(authorizationPolicy)
// check the amount of requests received within a certain time-period
.bean(receivedRequestFilter)
// return the current state of the task to the client. If the job is done,
// the response will also include a download link as wel as an MD5 hash to
// verify the correctness of the downloaded archive
.bean(ReturnArchiveRequestStatus.class)
.endRest();
ExtractArchiveRequestSettings class 仅对接收到的有效负载执行健全性检查并为缺失字段设置默认值。之后，请求被存储到数据库中，其唯一标识符存储到 header.
ArchiveRequestSetting 看起来确实像下面的示例（稍微简化）
@Getter
@Setter
@JsonIgnoreProperties(ignoreUnknown = true)
@JsonInclude(JsonInclude.Include.NON_NULL)
public class ArchiveRequestSettings {
/** Specifies if sent or received messages should be included in the artifact. Setting this field
* to 'DELIVERED' will include only delivered documents where the companyUuid of the requesting
* user matches the documents sender identifier. Specifying this field as 'RECEIVED' will include
* only documents whose receiver identifier matches the companyUuid of the requesting user. **/
private String direction;
/** The naming schema of entries within the archive **/
private String entryPattern;
/** The upper timestamp bound to include messages. Entries older than this value will be omitted **/
@JsonSerialize(using = Iso8601DateSerializer.class)
@JsonDeserialize(using = Iso8601DateDeserializer.class)
private Date from;
/** The lower timestamp bound to include messages. Entries younger than this value will be
* omitted. If left empty this will include even the most recent messages. **/
@JsonSerialize(using = Iso8601DateSerializer.class)
@JsonDeserialize(using = Iso8601DateDeserializer.class)
private Date till;
}
ReturnArchiveRequestCreatedStatus class 查找存储的请求实体并 return 使用 202 Accepted 响应对其进行处理。
@Handler
public void returnStatus(Exchange exchange) {
String archiveId = exchange.getIn().getHeader(HeaderConstants.ARCHIVES_REQUEST_ID, String.class);
ArchiveRequestEntity archive = repository.findOne(archiveId);
Message msg = new DefaultMessage(exchange.getContext());
msg.setHeader(Exchange.HTTP_RESPONSE_CODE, 202); // Accepted
msg.setHeader(Exchange.CONTENT_TYPE, "application/json; charset=\"utf-8\"");
msg.setHeader("Location", archiveLocationUrl + "/" + archiveId);
msg.setBody(archive);
exchange.setOut(msg);
}
返回存储请求的当前状态可确保客户端可以检查实际应用了哪些设置，并可以在某些默认设置不方便或需要应用进一步更改时更新它们。
实际的支持过程是通过将交换发送到 Redis queue 开始的，该 Redis queue 在不同的机器上使用。此过程的输出将是一个包含请求文件的存档，该文件将上传到 public 可访问位置，并且只有 link 将存储在请求实体中。请注意，我们有一个自定义的 camel 组件，它模仿了 Redis queues 的 seda entpoint。使用 seda 虽然应该足以在不同的线程中开始处理任务。
根据支持进程的当前状态，存储的请求实体将由支持进程更新。在收到状态请求（通过 GET）时，查询数据存储的当前状态并映射到某些响应：
public class ReturnArchiveRequestStatus {
@Resource
private ArchiveRequestRepository repository;
@Handler
public void returnArchiveStatus(Exchange exchange) throws JSONException {
String archiveId = exchange.getIn().getHeader("archiveId", String.class);
if (StringUtils.isBlank(archiveId)) {
badRequest(exchange);
return;
}
ArchiveRequestEntity archive = repository.findOne(archiveId);
if (null == archive) {
notFound(archiveId, exchange);
return;
}
ok(archive, exchange);
}
private void badRequest(Exchange exchange) throws JSONException {
Message msg = new DefaultMessage(exchange.getContext());
msg.setHeader(Exchange.HTTP_RESPONSE_CODE, 400);
msg.setHeader(Exchange.CONTENT_TYPE, "application/json; charset=\"utf-8\"");
msg.setFault(false);
JSONObject json = new JSONObject();
json.put("status", "ERROR");
json.put("message", "No archive identifier found");
msg.setBody(json.toString());
exchange.setOut(msg);
}
private void notFound(String archiveId, Exchange exchange) throws JSONException {
Message msg = new DefaultMessage(exchange.getContext());
msg.setHeader(Exchange.HTTP_RESPONSE_CODE, 403);
msg.setHeader(Exchange.CONTENT_TYPE, "application/json; charset=\"utf-8\"");
msg.setFault(false);
JSONObject json = new JSONObject();
json.put("status", "ERROR");
json.put("message", "Could not find pending archive process with ID " + archiveId);
msg.setBody(json.toString());
exchange.setOut(msg);
}
private void ok(UserArchiveRequestEntity archive, Exchange exchange) throws JSONException {
Message msg = new DefaultMessage(exchange.getContext());
msg.setHeader(Exchange.HTTP_RESPONSE_CODE, 200);
msg.setHeader(Exchange.CONTENT_TYPE, "application/json; charset=\"utf-8\"");
msg.setFault(false);
msg.setBody(archive);
exchange.setOut(msg);
}
}
在整个过程中存储和更新的实际实体看起来是这样的（简化）：
@Getter
@Setter
@Builder
@ToString
@Document(collection = "archive")
@JsonInclude(JsonInclude.Include.NON_EMPTY)
public class ArchiveRequestEntity {
/**
* The current state of the archiving process
*/
public enum State {
/** The request to create an archive was cued but not yet processed **/
QUEUED,
/** The archive is currently under construction **/
RUNNING,
/** The archive was generated successfully. {@link #downloadUrl} should contain the link the
* archive can be found **/
FINISHED,
/** Indicates that the archive generation failed. {@link #error} should indicate the actual
* reason why the request failed **/
FAILED
}
@Id
@JsonIgnore
private String id;
/** Timestamp the process was triggered **/
@JsonIgnore
@Indexed(expireAfterSeconds = DEFAULT_EXPIRE_TIME)
private Date timestamp = new Date();
/** The identifier of the company to create the archive for **/
private String companyUuid;
/** The state this archive is currently in **/
private State state = State.QUEUED;
...
/** Marks the upper limit to include entries to the archive. Entries older then this field will
* not be included in the archives while entries equal or younger than this timestamp will be
* included unless they are younger than {@link #till} timestamp **/
@JsonFormat(pattern = "yyyy-MM-dd'T'HH:mm:ssXX")
private Date from;
/** Marks the lower limit to include entries to the archive. Entries younger than this field will
* not be included in the archive **/
@JsonFormat(pattern = "yyyy-MM-dd'T'HH:mm:ssXX")
private Date till;
/** Information on why the archive creation failed **/
private String error;
/** The URL of the final archive to download **/
private String downloadUrl;
/** The MD5 Hash of the final artifact in order to guarantee clients an unmodified version of the
* archive **/
private String md5Hash;
...
}
请注意，无论处理状态的当前状态如何，200 OK 都将 return 与当前的 JSON 处理状态表示形式相结合。客户端将看到具有 downloadUrl 和 md5Hash 属性集的 FINISHED 状态，或者看到具有不同属性的不同状态。
当然，支持进程需要适当地更新请求状态，否则客户端将无法检索到有关请求当前状态的正确信息。
这种方法应该适用于几乎所有长 运行 进程，尽管您传递的信息的内部结构可能与我们的场景不同。希望这有帮助