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&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&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&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();
}
}

 
以上代码经测试，都可运行。