camel系列-声明标签(Claim Check)

发表于 更新于

概念

消息中缺少所需的数据项时，可以使用内容扩充器加以补充。反之，如果要将消息中的多余数据项删除，可以使用内容过滤器。但是，我们可能只是临时删除这些数据项。例如，消息中包含了后续消息流要用到的数据项，但是所有中间处理步骤都不会使用这些数据项。因此，为了避免性能受到影响，也为了降低调试的难度，我们没有必要在中间处理步骤中携带这些额外的数据。

在不损害信息内容的前提条件下，怎样减少跨系统发送的消息的数据量？

把消息中的数据存储在一个持久存储库中，并把声明标签传递给后续组件。这些组件可以使用声明标签重新获得所存储的信息.

流程步骤

声明标签模式包括以下步骤 ∶

  1. 带有数据的消息到达。
  2. 托运行李(Check Luggage)组件为信息产生唯一的键。该键可以作为后续的声明标签。
  3. 托运行李组件从消息中取出数据，并把数据存储在一个持久的存储库中，如文件或数据存储库。它把键与所存储的数据相关联。
  4. 从消息中删除持久存储的数据，并在消息中增加声明标签。
  5. 其他组件可以使用内容扩充器，根据声明标签取得数据。

原型示例

  • checkLuggage 负责 1-3 的步骤流程
  • addDataBackIn 负责 4-5 的步骤流程
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
public class ClaimCheckTest extends ContextTestSupport {

    // in memory data store for testing only!
    public static Map<String, Object> dataStore = new HashMap<>();

    @Test
    public void testClaimCheck() throws Exception {
        String body = "<order custId=\"123\"><lotsOfContent/></order>";

        // check to make sure the message body gets added back in properly
        MockEndpoint resultEndpoint = getMockEndpoint("mock:result");
        resultEndpoint.expectedMessageCount(1);
        resultEndpoint.message(0).body().isEqualTo(body);

        // check to make sure the claim check is added to the message and
        // the body is removed
        MockEndpoint testCheckpointEndpoint = getMockEndpoint("mock:testCheckpoint");
        testCheckpointEndpoint.expectedMessageCount(1);
        testCheckpointEndpoint.expectedHeaderReceived("claimCheck", "123");
        testCheckpointEndpoint.message(0).body().isNull();

        template.sendBody("direct:start", body);

        assertMockEndpointsSatisfied();
    }

    @Override
    protected Registry createRegistry() throws Exception {
        Registry jndi = super.createRegistry();
        jndi.bind("checkLuggage", new CheckLuggageBean());
        jndi.bind("dataEnricher", new DataEnricherBean());
        return jndi;
    }

    @Override
    protected RouteBuilder createRouteBuilder() {
        return new RouteBuilder() {
            public void configure() {
                // START SNIPPET: e1
                from("direct:start").to("bean:checkLuggage",
										"mock:testCheckpoint",
										"bean:dataEnricher",
										"mock:result");
                // END SNIPPET: e1
            }
        };
    }

    // START SNIPPET: e2
    public static final class CheckLuggageBean {
        public void checkLuggage(Exchange exchange, @Body String body, @XPath("/order/@custId") String custId) {
            // store the message body into the data store, using the custId as
            // the claim check
            dataStore.put(custId, body);
            // add the claim check as a header
            exchange.getIn().setHeader("claimCheck", custId);
            // remove the body from the message
            exchange.getIn().setBody(null);
        }
    }
    // END SNIPPET: e2

    // START SNIPPET: e3
    public static final class DataEnricherBean {
        public void addDataBackIn(Exchange exchange, @Header("claimCheck") String claimCheck) {
            // query the data store using the claim check as the key and add the
            // data
            // back into the message body
            exchange.getIn().setBody(dataStore.get(claimCheck));
            // remove the message data from the data store
            dataStore.remove(claimCheck);
            // remove the claim check header
            exchange.getIn().removeHeader("claimCheck");
        }
    }
    // END SNIPPET: e3
}

claimCheck 方法

  1. claimCheck 操作

使用此 EIP 时，您必须指定要使用的操作，可以是以下各项：

  • Get - 通过给定键获取（不删除）声明检查。
  • GetAndRemove - 通过给定键获取和删除声明检查。
  • Set - 使用给定的密钥设置一个新的（如果密钥已经存在，将覆盖）声明检查。
  • Push - 在堆栈上设置新的声明检查（不使用密钥）。
  • Pop - 从堆栈中获取最新的声明检查（不使用密钥）。

使用 Get、GetAndRemove 或 Set 操作时，您必须指定一个键。然后，这些操作将使用此密钥存储和检索数据。您可以使用它在不同的键中存储多个数据。

PushandPop 操作不使用键，而是将数据存储在堆栈结构中。

  1. claimCheck 方法示例
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
public class ClaimCheckEipGetSetTest extends ContextTestSupport {

    @Test
    public void testGetSet() throws Exception {
        getMockEndpoint("mock:a").expectedBodiesReceived("Hello World");
        getMockEndpoint("mock:b").expectedBodiesReceived("Bye World");
        getMockEndpoint("mock:c").expectedBodiesReceived("Hello World");
        getMockEndpoint("mock:d").expectedBodiesReceived("Hi World");
        getMockEndpoint("mock:e").expectedBodiesReceived("Hello World");

        template.sendBody("direct:start", "Hello World");

        assertMockEndpointsSatisfied();
    }

    @Override
    protected RouteBuilder createRouteBuilder() throws Exception {
        return new RouteBuilder() {
            @Override
            public void configure() throws Exception {
                from("direct:start").to("mock:a")
                    .claimCheck(ClaimCheckOperation.Set, "foo")
                    .transform().constant("Bye World")
                        .to("mock:b")
                        .claimCheck(ClaimCheckOperation.Get, "foo").to("mock:c")
                    .transform().constant("Hi World").to("mock:d")
                        .claimCheck(ClaimCheckOperation.Get, "foo").to("mock:e");
            }
        };
    }
}

Push 和 Pop

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
public class ClaimCheckEipPushPopBodyTest extends ContextTestSupport {

    @Test
    public void testPushPopBody() throws Exception {
        getMockEndpoint("mock:a").expectedBodiesReceived("Hello World");
        getMockEndpoint("mock:a").expectedHeaderReceived("foo", 123);
        getMockEndpoint("mock:b").expectedBodiesReceived("Bye World");
        getMockEndpoint("mock:b").expectedHeaderReceived("foo", 456);
        getMockEndpoint("mock:c").expectedBodiesReceived("Hello World");
        getMockEndpoint("mock:c").expectedHeaderReceived("foo", 456);

        template.sendBodyAndHeader("direct:start", "Hello World", "foo", 123);

        assertMockEndpointsSatisfied();
    }

    @Override
    protected RouteBuilder createRouteBuilder() throws Exception {
        return new RouteBuilder() {
            @Override
            public void configure() throws Exception {
                from("direct:start").to("mock:a").
					claimCheck(ClaimCheckOperation.Push).transform().constant("Bye World")
                        .setHeader("foo", constant(456)).to("mock:b")
                        // only merge in the message body
                        .claimCheck(ClaimCheckOperation.Pop, null, "body").to("mock:c");
            }
        };
    }
}

filter

只想取回 header 名为 foo 或 bar 的数据

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
<route>
	<from uri="direct:start"/>
	<to uri="mock:a"/>
	<claimCheck operation="Push"/>
	<transform>
		<constant>Bye World</constant>
	</transform>
	<setHeader name="foo">
		<constant>456</constant>
	</setHeader>
	<removeHeader headerName="bar"/>
	<to uri="mock:b"/>
	<!-- only merge in the message headers foo or bar -->
	<claimCheck operation="Pop" filter="header:(foo|bar)"/>
	<to uri="mock:c"/>
</route>