Spring integration总结


1、概念

  可以理解成一个队列,在该队列不同节点部分进行相应的逻辑操作,实现轻量级消息传递

 2、结构流程

 3、代码结构:

  3.1配置文件:

<?xml version="1.0" encoding="UTF-8"?>

       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:integration="http://www.springframework.org/schema/integration"
       xmlns:beans="http://www.springframework.org/schema/beans"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">

    

    package="ch.javaee.integration.example.helloWorld"/>

    
    

    

    
    

    

    
    
    

    
    
    

    

             

 其中1、2、3、4都是定义的channel 

 5、6、7是针对各个channel中数据处理,ref是处理input-channel数据的bean名称,method是ref中bean里的方法,output-channel存储是处理后的结果

 Service Activator:可调用Spring的Bean来处理消息,并将处理后的结果输出到指定的消息通道,其作用使用起来和在配置文件中的method="****"类似

3.2  channel数据处理逻辑

以5为例:

   3.2.1 input-channel内数据内容: 

public class MyTestDemo {

    public static void main(String[] args) {
        MyTestProducerService myTestService=new MyTestProducerService();
        // create a message with the content "World"
//        buildMessageAndSend("notification", "identify");
        List stringList=new ArrayList();
        stringList.add("Monday");
        stringList.add("Tuesday");
        stringList.add("Wednesday");
        for(int i=0;i){
            myTestService.putMessage(stringList.get(i));
        }
        buildMessageAndSend("notification", "shutdown");
    }


    private static Map buildMessageMap(String type, String msg) {
        Map msgMap = new HashMap();
        msgMap.put("type", type);
        msgMap.put("message", msg);
        msgMap.put("context_id", "123");
        return msgMap;
    }

    private static void buildMessageAndSend(String type, String msg) {
        Map notifyMap = buildMessageMap(type, msg);
        // load the Spring context
        ApplicationContext context = new ClassPathXmlApplicationContext("my-spring-config.xml");
        // get the reference to the message channel
        MessageChannel channel = context.getBean("inputChannel", MessageChannel.class);
        try {
            if(channel.send(MessageBuilder.withPayload(notifyMap).build()) == false) {
                throw new Exception("Unable to publish the shutdown notification.");
            }
        } catch (Exception e) {
            System.out.println(e);
        }
    }
}

   其中 1:加载配置文件

      2:获取输入channel名字

      3:发送数据到input-channel中

 3.2.2 处理input-channel中数据

@Service
public class MyTestProducerService {
    /** The message queue. */
    private static BlockingQueue messageQueue = new ArrayBlockingQueue(
            100, true);

    public static void putMessage(Object obj){
        messageQueue.add(obj);
    }

    public final Message poll() throws InterruptedException {
        Object obj = null;
        obj=messageQueue.take();
        Message message = null;
        if(obj instanceof String){
            String objStr=(String) obj;
            message= MessageBuilder.withPayload((Object)objStr)
                    .build();
        }
        return message;
    }
}


   其中1:对应3.1节中5中ref对应的bean

    2:对应3.1节中5中method对应的方法

    3:返回给3.1节中5中output-channel中的数据

3.3 其他channel处理逻辑与3.2节类似了