STORM入门之（集成Redis）

首先需要Redis工具类，自己写了一个简易的源码地址：http://blog.csdn.net/yl3395017/article/details/76172598

测试数据 KEY:test  VALUE:world

POM

<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-redis</artifactId>
<version>1.0.0</version>
<type>jar</type>
</dependency>

RedisBolt

package com.storm.redis;

import org.apache.storm.redis.bolt.AbstractRedisBolt;
import org.apache.storm.redis.common.config.JedisPoolConfig;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import redis.clients.jedis.JedisCommands;

/**
* Created with IntelliJ IDEA.
* User: Administrator
* Date: 17-8-23
* Time: 下午2:00
* To change this template use File | Settings | File Templates.
*/
public class RedisBolt extends AbstractRedisBolt {
public RedisBolt(JedisPoolConfig config) {
super(config);
}

@Override
public void execute(Tuple input) {
JedisCommands jedisCommands = null;
try {
jedisCommands = getInstance();
String wordName = input.getStringByField("index");
String countStr = jedisCommands.get(wordName);
if (countStr != null) {
System.out.print("#######countStr#########:"+countStr);
} else {

}
} finally {
if (jedisCommands != null) {
returnInstance(jedisCommands);
}
}
}

@Override
public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
//To change body of implemented methods use File | Settings | File Templates.
}
}

Topology构建方式

/**
* 构建RedisBolt
*/
private static void builtRedisBolt(TopologyBuilder builder){
JedisPoolConfig poolConfig = new JedisPoolConfig.Builder().setHost("10.2.4.12").setPort(6379).build();
RedisBolt rb = new  RedisBolt(poolConfig);
builder.setBolt("redis-bolt",rb,1).shuffleGrouping("BoltA");
}

结果