Apache Camel的aggregate简单示例

约定不等于承诺〃 2022-08-04 15:42 128阅读 0赞

aggregate表示聚合，把相同类型的对象聚集起来，形成一个新的对象返回出去

下面的例子演示了一个聚合求用户总得分的例子

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
    	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:camel="http://camel.apache.org/schema/spring"
    	xsi:schemaLocation="
            http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
            http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">
    
    	<bean id="echo" class="com.lala.bean.Echo"></bean>
    	<bean id="scoreAggregationStrategy" class="com.lala.bean.ScoreAggregationStrategy" />
    
    	<camelContext id="camelContext" xmlns="http://camel.apache.org/schema/spring">
    		<template id="camelContextTemplate"/>
    		<route>
    			<from uri="direct:start"/>
    			
    			<aggregate strategyRef="scoreAggregationStrategy" completionSize="3">
    				<correlationExpression>
    					
    					<simple>in.header.name</simple>
    				</correlationExpression>
    				
    				<completionPredicate>
    	                <simple>${in.header.size} == 3</simple>
    	            </completionPredicate>
    				<removeHeader headerName="course"></removeHeader>
    				<removeHeader headerName="score"></removeHeader>
    				<to uri="bean:echo" />
    			</aggregate>
    		</route>
    	</camelContext>
    	
    </beans>

注意：aggregate的 completionSize="3"  和  completionPredicate 只能有一个

package com.lala.bean;
    
    import org.apache.camel.Exchange;
    import org.apache.camel.processor.aggregate.AggregationStrategy;
    
    public class ScoreAggregationStrategy implements AggregationStrategy 
    {
    	Integer getIntValueFromHeader(Exchange exch, String key)
    	{
    		return exch.getIn().getHeader(key, Integer.class);
    	}
    	/**
    	 * 第一次的时候，oldExchange为null
    	 * 以后，newExchange都是下一次的oldExchange
    	 */
    	public Exchange aggregate(Exchange oldExchange, Exchange newExchange)
    	{
    		if(oldExchange == null)
    		{
    			newExchange.getIn().setHeader("total", getIntValueFromHeader(newExchange, "score"));
    			newExchange.getIn().setHeader("size", 1);
    		}
    		else
    		{
    			newExchange.getIn().setHeader("total", getIntValueFromHeader(oldExchange, "total") + getIntValueFromHeader(newExchange, "score"));
    			newExchange.getIn().setHeader("size", getIntValueFromHeader(oldExchange, "size") + 1);
    		}
    		return newExchange;
    	}
    }

package com.lala.bean;
    
    import org.apache.camel.Exchange;
    import org.apache.camel.Processor;
    
    public class Echo implements Processor 
    {
    	public void process(Exchange exchange) throws Exception 
    	{
    		System.out.println(exchange.getIn().getHeaders());
    	}
    
    }

最后调用：

package com.lala.camel;
    
    import java.util.ArrayList;
    import java.util.Collections;
    import java.util.List;
    import java.util.concurrent.TimeUnit;
    
    import org.apache.camel.Exchange;
    import org.apache.camel.Processor;
    import org.apache.camel.ProducerTemplate;
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    
    public class CamelTest 
    {
    	public static void main(String[] args)throws Exception
    	{
    		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("applicationContext3.xml");
    			
    		ProducerTemplate pt = context.getBean("camelContextTemplate", ProducerTemplate.class);
    		
    		List<User> users = new ArrayList<>();
    		users.add(new User("刘星", "语文", 80));
    		users.add(new User("刘星", "数学", 70));
    		users.add(new User("刘星", "外语", 65));
    		
    		users.add(new User("夏雨", "语文", 65));
    		users.add(new User("夏雨", "数学", 60));
    		users.add(new User("夏雨", "外语", 40));
    		
    		users.add(new User("夏雪", "语文", 90));
    		users.add(new User("夏雪", "数学", 60));
    		users.add(new User("夏雪", "外语", 80));
    		
    		Collections.shuffle(users);
    		
    		for(final User user : users)
    		{
    			pt.send("direct:start", new Processor(){
    				public void process(Exchange exchange) throws Exception {
    					exchange.getIn().setHeader("name", user.getName());
    					exchange.getIn().setHeader("course", user.getCourse());
    					exchange.getIn().setHeader("score", user.getScore());
    				}
    			});
    		}
    
    		TimeUnit.SECONDS.sleep(20);
    		context.close();
    	}
    }
    class User
    {
    	private String name;
    	private String course;	//科目
    	private Integer score;	//得分
    	public User(String name, String course, Integer score) {
    		this.name = name;
    		this.course = course;
    		this.score = score;
    	}
    	public String getName() {
    		return name;
    	}
    	public void setName(String name) {
    		this.name = name;
    	}
    	public String getCourse() {
    		return course;
    	}
    	public void setCourse(String course) {
    		this.course = course;
    	}
    	public Integer getScore() {
    		return score;
    	}
    	public void setScore(Integer score) {
    		this.score = score;
    	}
    }

运行，输出结果如下：

\{breadcrumbId=ID-html-65231-1441381869323-0-9, name=夏雨, size=3, total=165\}  
\{breadcrumbId=ID-html-65231-1441381869323-0-15, name=夏雪, size=3, total=230\}  
\{breadcrumbId=ID-html-65231-1441381869323-0-21, name=刘星, size=3, total=215\}