一哥一射
问题描述
我有一个用例来并行读取员工、地址和联系人文件并将其转换为 beanIO 对象并合并 beanIO 对象以生成完整的employeeDetails 对象.
I have the usecase to read employee, address and contact files in parallel and convert that to beanIO object and merge the beanIO object to produce the complete employeeDetails object.
Emp 文件:
1 Foo Engineer 2 Bar AssistantEngineer
Emp 联系文件:
1 8912345678 foo@org.com 2 7812345678 bar@org.com
Emp 地址文件:
1 city1 1234 2 city2 2345
Exchange 中 EmployeeDetailsBeanIODataFormat 对象的预期输出:
Expected output in a EmployeeDetailsBeanIODataFormat object in Exchange:
1 Foo Engineer foo@org.com city1 1234 2 Bar AssistantEngineer bar@org.com city2 2345
我有以下路线
from("file://C:/cameltest/employee.txt").to("seda:beanIO");
from("file://C:/cameltest/employeeContact.txt").to("seda:beanIOContact");
from("file://C:/cameltest/employeeAddress.txt").to("seda:beanIOAddress");
每个文件都转换为beanio对象
Each file is converted to beanio object
BeanIODataFormat empFormat = new BeanIODataFormat("beanIO.xml","emp");
BeanIODataFormat empContactFormat = new BeanIODataFormat("beanIO.xml", "empContact");
BeanIODataFormat empAddressFormat = new BeanIODataFormat("beanIO.xml", "empAddress");
from("seda:beanIO").unmarshal(empFormat).log("body - ${body}");
from("seda:beanIOContact").unmarshal(empContactFormat).log("Contact body ${body}");
from("seda:beanIO").unmarshal(empAddressFormat).log("Address body - ${body}");
输出正确记录了 bean 对象.
The output logs the bean objects correctly.
现在我需要合并对象以形成 EmployeeDetails 对象.有人可以让我知道该怎么做吗?我读过，似乎聚合器可以用来做这项工作，但不确定这种方法.
Now I need to merge the objects to form EmployeeDetails object. Can someone let me know how to do this? I have read and it seems like Aggregators can be used to do this job, but not sure with the approach.
有关此示例的任何想法都会有所帮助.欢迎提出建议，是否建议先根据员工 ID 合并文件并从中创建一个对象?在这种情况下，我不想将合并的文件写入磁盘，因为 IO 会降低性能.
Any idea on this with sample will be helpful. Suggestions are welcome, is it advisable to merge the files first based on employee id and create an object out of it? I dont want to write the merged file into disk in this case as IO will cost the performance.
提前致谢.
推荐答案
解组后使用拆分器拆分每条消息
Use a splitter to split each message after the unmarshalling
from("seda:beanIO").unmarshal(empFormat).split(body()).to("seda:aggregate");
from("seda:beanIOContact").unmarshal(empContactFormat).split(body()).to("seda:aggregate");
from("seda:beanIOAddress").unmarshal(empAddressFormat).split(body()).to("seda:aggregate");
这就是聚合器的样子.详细信息对象作为标头存储在 olddExchange 中.最重要的参数如下
And then here is how an aggregator could look like. The details object is stored as a header in the olddExchange. The most important parameters are the following
- correlationExpression: simple("${body.id}") 关联具有相同 id(1 或 2)的所有消息
- completionSize=3.每个文件一个.
<小时>
from("seda:aggregate").aggregate(simple("${body.id}"), (oldExchange,newExchange) -> {
if (oldExchange == null) {
EmployeeDetails details = buildDetails(new EmployeeDetails(), newExchange);
newExchange.getIn().setHeader("details", details);
return newExchange;
}
EmployeeDetails details = oldExchange.getIn().getHeader("details", EmployeeDetails.class);
buildDetails(details, newExchange);
oldExchange.getIn().setHeader("details", details);
return oldExchange;
}).completionSize(3).log("Details - ${header.details}")
还有
private EmployeeDetails buildDetails(EmployeeDetails details, Exchange newExchange) {
Object newBody = newExchange.getIn().getBody();
if (newBody instanceof Employee) {
details.setId(((Employee) newBody).getId());
details.setName(((Employee) newBody).getName());
details.setJob(((Employee) newBody).getJob());
} else if (newBody instanceof EmployeeContact) {
details.setEmail(((EmployeeContact) newBody).getEmail());
} else if (newBody instanceof EmployeeAddress) {
details.setCity(((EmployeeAddress) newBody).getCity());
details.setCode(((EmployeeAddress) newBody).getCode());
}
return details;
}
那么结果将是 2 个细节对象
Then the outcome would be 2 details objects
Details - EmployeeDetails(id=1, name=Foo, job=Engineer, email=foo@org.com, city=city1, code=1234) Details - EmployeeDetails(id=2, name=Bar, job=AssistantEnginee, email=bar@org.com, city=city2, code=2345)
