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();
    }
}
复制代码

 

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

 

 

 
 
转自:https://www.cnblogs.com/zengbiaobiao2016/p/5481029.html