Apache Camel系列(3)----Redis組件

Redis組件允許你從Redis接收消息，以及將消息發送給Redis。RedisProducer的功能很強大，幾乎能執行所有的Redis Command，這些Command都是在Message的header中進行設置的。遺憾的是RedisConsumer僅僅支持pub/sub模式，不支持Point2Point，這意味這在Camel中，通過阻塞的方式消費Lists中的消息是不可行的。我反饋了這個問題到Apache Camel Mail List，希望以后的版本支持P2P更能。下面演示如何使用camel-spring-redis組件。

 

使用Spring

1，創建Maven工程，添加camel-spring-redis引用，pom.xml文件內容如下：

 

<dependencies>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>3.8.1</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-core</artifactId>
            <version>2.17.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-spring</artifactId>
            <version>2.17.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-spring-redis</artifactId>
            <version>2.17.0</version>
        </dependency>
    </dependencies>

2，在資源文件下下創建spring bean配置文件，內容如下:

 

<?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="connectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory">
        <property name="hostName" value="localhost" />
        <property name="port" value="9999" />
        <property name="password" value="1234567890" />
    </bean>
    <bean id="serializer" class="org.springframework.data.redis.serializer.StringRedisSerializer" />

    <camelContext id="camel" xmlns="http://camel.apache.org/schema/spring">
        <route startupOrder="1">
            <from uri="timer://foo?fixedRate=true&amp;period=1000"/>
            <setHeader headerName="CamelRedis.Command">
                <constant>SET</constant>
            </setHeader>
            <setHeader headerName="CamelRedis.Key">
                <constant>keyOne</constant>
            </setHeader>
            <setHeader headerName="CamelRedis.Value">
                <constant>valueOne</constant>
            </setHeader>
            <to uri="spring-redis://localhost:9999?connectionFactory=#connectionFactory&amp;serializer=#serializer"/>
        </route>
    </camelContext>
</beans>

上邊的beans文件中，定義了bean connectionFactory，可以設置redis服務器的相關信息，比如密碼。如果你的redis服務器沒有設置密碼，那么這個bean定義可以省略，此時配置文件如下:

 

<?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="serializer" class="org.springframework.data.redis.serializer.StringRedisSerializer" />

    <camelContext id="camel" xmlns="http://camel.apache.org/schema/spring">
        <route startupOrder="1">
            <from uri="timer://foo?fixedRate=true&amp;period=1000"/>
            <setHeader headerName="CamelRedis.Command">
                <constant>SET</constant>
            </setHeader>
            <setHeader headerName="CamelRedis.Key">
                <constant>keyOne</constant>
            </setHeader>
            <setHeader headerName="CamelRedis.Value">
                <constant>valueOne</constant>
            </setHeader>
            <to uri="spring-redis://localhost:9999?serializer=#serializer"/>
        </route>
    </camelContext>
</beans>

setHeader中設置了Redis的相關命令，RedisProducer支持幾乎所有的RedisCmmand，具體可參見 Redis Component官網。

這里的意思是每隔1S為Redis服務器的keyOnve設置一個值valueOne。

 

3， 創建App3.java，啟動spring。

 

/**
 * Created by sam on 5/10/16.
 */
public class App3 {
    public static void main(String[] args) throws Exception {
        ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("context.xml");
        context.start();
        System.in.read();
    }
}

啟動ClassPathXmlApplicationContext的時候，系統會自動創建並運行CamelContext。

 

監控Redis，得到結果如下：

 

[sam@localhost redis-2.8.19]$ src/redis-cli -p 9999 -a 1234567890
127.0.0.1:9999> MONITOR
OK
1462931627.929228 [0 127.0.0.1:50939] "PING"
1462931686.110952 [0 127.0.0.1:50943] "AUTH" "1234567890"
1462931686.120240 [0 127.0.0.1:50943] "SET" "keyOne" "valueOne"
1462931687.065705 [0 127.0.0.1:50943] "SET" "keyOne" "valueOne"
1462931688.066442 [0 127.0.0.1:50943] "SET" "keyOne" "valueOne"
1462931689.066169 [0 127.0.0.1:50943] "SET" "keyOne" "valueOne"
1462931690.065948 [0 127.0.0.1:50943] "SET" "keyOne" "valueOne"
1462931691.065674 [0 127.0.0.1:50943] "SET" "keyOne" "valueOne"

 

不使用Spring

公司不是所有的項目都用了Spring，所以研究了兩天，做了個非Spring的Demo，還是maven工程，pom.xml配置文件和上邊的一樣，創建app4.java類，代碼如下:

 

 

public class App4 {
    public static void main(String[] args) throws Exception {
        JedisConnectionFactory connectionFactory = new JedisConnectionFactory(); // 創建connectionFactory
        connectionFactory.setHostName("localhost");
        connectionFactory.setPassword("1234567890");
        connectionFactory.setPort(9999);
        SimpleRegistry registry = new SimpleRegistry();        
        connectionFactory.afterPropertiesSet(); // 必須要調用該方法來初始化connectionFactory
        registry.put("connectionFactory", connectionFactory); //注冊connectionFactory
        registry.put("serializer", new StringRedisSerializer()); //注冊serializer

        CamelContext context = new DefaultCamelContext(registry);
        context.addRoutes(new RouteBuilder() {
            public void configure() {
                errorHandler(deadLetterChannel("stream:out"));
                from("timer://foo?fixedRate=true&period=1000").
                        setHeader("CamelRedis.Command", constant("PUBLISH")).
                        setHeader("CamelRedis.Channel", constant("testChannel")).
                        setHeader("CamelRedis.Message", constant(new Date().toString())).
                        to("spring-redis://localhost:9999?connectionFactory=#connectionFactory&serializer=#serializer");
            }
        });
        context.setTracing(true);
        context.start();
        Thread.sleep(Integer.MAX_VALUE);
        context.stop();
    }
}

 

這段代碼主要是使用SimpleRegistry來注冊bean信息，並將SimpleRegistry作為CamelContext的參數，這樣在endpoint的Url中就可以使用之前注冊的bean了。

另外在創建完connectionFactory后要調用afterPropertiesSet()方法來完成初始化。如果你的redis沒有設置密碼，並且不需要serializer，那么代碼更簡單，如下:

 

public class App4 {
    public static void main(String[] args) throws Exception {
        CamelContext context = new DefaultCamelContext();
        context.addRoutes(new RouteBuilder() {
            public void configure() {
                errorHandler(deadLetterChannel("stream:out"));
                from("timer://foo?fixedRate=true&period=1000").
                        setHeader("CamelRedis.Command", constant("PUBLISH")).
                        setHeader("CamelRedis.Channel", constant("testChannel")).
                        setHeader("CamelRedis.Message", constant(new Date().toString())).
                        to("spring-redis://localhost:9999");
            }
        });
        context.setTracing(true);
        context.start();
        Thread.sleep(Integer.MAX_VALUE);
        context.stop();
    }
}

 

以上代碼經測試，都可運行。