Storm如何接收数据

共计 4769 个字符，预计需要花费 12 分钟才能阅读完成。

这篇文章主要讲解了“Storm 如何接收数据”，文中的讲解内容简单清晰，易于学习与理解，下面请大家跟着丸趣 TV 小编的思路慢慢深入，一起来研究和学习“Storm 如何接收数据”吧！

    简要的模拟如何接收数据：

package com.cc.storm.spout;
import java.io.IOException;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.LinkedBlockingQueue;
import org.apache.log4j.Logger;
import redis.clients.jedis.JedisPubSub;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Utils;
public class RandomEmitSpout extends BaseRichSpout {
 private Random _random;
 private static final long serialVersionUID = 4092527421163270357L;
 static Logger LOG = Logger.getLogger(RandomEmitSpout.class);
 private SpoutOutputCollector _collector;
 @Override
 public void open(Map conf, TopologyContext context,
 SpoutOutputCollector collector) {
 _collector = collector;
 _random = new Random();
 @Override
 public void nextTuple() {
 try {Thread.sleep(1000);
 } catch (Exception e) {e.printStackTrace();
 String[] userIds = {  1 ,  2 ,  3 ,  4  };
 String[] merchandiseIDS = {  1  };
 _collector.emit(new Values(userIds[_random.nextInt(userIds.length)],
 merchandiseIDS[_random.nextInt(merchandiseIDS.length)]));
 @Override
 public void declareOutputFields(OutputFieldsDeclarer declarer) {
 // TODO Auto-generated method stub
 declarer.declare(new Fields( userIdS ,  merchandiseIDS));
 @Override
 public void close() {}

     plus：如果您采用的是 Redis

    那么：   

package com.cc.storm.spout;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import org.apache.log4j.Logger;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.JedisPubSub;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Utils;
public class RedisPubSubSpout extends BaseRichSpout {
  * @Fields serialVersionUID : TODO
  */
 private static final long serialVersionUID = 4092527421163270357L;
 static Logger LOG = Logger.getLogger(RedisPubSubSpout.class);
 private SpoutOutputCollector _collector;
 private final String host;
 private final int port;
 private final String pattern;
 LinkedBlockingQueue String  queue;
 JedisPool pool;
 public RedisPubSubSpout(String host, int port, String pattern) {
 // TODO Auto-generated constructor stub
 this.host = host;
 this.port = port;
 this.pattern = pattern;
 //  监听线程，从 redis 订阅的兴趣事件中获取数据
 class ListenerThread extends Thread {
 private LinkedBlockingQueue String  queue;
 JedisPool pool;
 String pattern;
 public ListenerThread(LinkedBlockingQueue String  queue,
 JedisPool pool, String pattern) {
 // TODO Auto-generated constructor stub
 this.queue = queue;
 this.pool = pool;
 this.pattern = pattern;
 @Override
 public void run() {JedisPubSub listener = new JedisPubSub() {
 @Override
 public void onUnsubscribe(String arg0, int arg1) {
 // TODO Auto-generated method stub
 @Override
 public void onSubscribe(String arg0, int arg1) {
 // TODO Auto-generated method stub
 @Override
 public void onPUnsubscribe(String arg0, int arg1) {
 // TODO Auto-generated method stub
 @Override
 public void onPSubscribe(String arg0, int arg1) {
 // TODO Auto-generated method stub
 @Override
 public void onPMessage(String pattern, String channel,
 String message) {
 // TODO Auto-generated method stub
 queue.offer(message);
 @Override
 public void onMessage(String channel, String message) {
 // TODO Auto-generated method stub
 queue.offer(message);
 Jedis jedis = pool.getResource();
 try {jedis.psubscribe(listener, pattern);
 } finally {pool.returnResource(jedis);
 @SuppressWarnings(rawtypes)
 @Override
 public void open(Map conf, TopologyContext context,
 SpoutOutputCollector collector) {
 // TODO Auto-generated method stub
 _collector = collector;
 //  队列最大支持 1000 个
 queue = new LinkedBlockingQueue String (1000);
 JedisPoolConfig config = new JedisPoolConfig();
 // error
 pool = null;
 ListenerThread listener = new ListenerThread(queue, pool, pattern);
 //  启动线程
 listener.start();
 @Override
 public void nextTuple() {
 // TODO Auto-generated method stub
 String ret = queue.poll();
 if (null == ret) {
 //  如果队列中暂无数据可取，休息 500ms
 Utils.sleep(500);
 } else {
 //  数据格式为  “userID:merchandiseID”，可以依据需求更改此处
 String[] s = ret.split( : 
 _collector.emit(new Values(s[0], s[1]));
 @Override
 public void declareOutputFields(OutputFieldsDeclarer declarer) {
 // TODO Auto-generated method stub
 declarer.declare(new Fields( userIdS ,  merchandiseIDS));
 @Override
 public void close() {
 // TODO Auto-generated method stub
 pool.destroy();}

感谢各位的阅读，以上就是“Storm 如何接收数据”的内容了，经过本文的学习后，相信大家对 Storm 如何接收数据这一问题有了更深刻的体会，具体使用情况还需要大家实践验证。这里是丸趣 TV，丸趣 TV 小编将为大家推送更多相关知识点的文章，欢迎关注！