文档章节

Storm[TOPN -排序] - RandomEmitSPout

止静
 止静
发布于 2014/09/03 17:56
字数 549
阅读 357
收藏 0

 

本章主题:

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

        

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

     simple Tips , 太简单,也不做过多的解释。

© 著作权归作者所有

止静
粉丝 121
博文 134
码字总数 125762
作品 0
东城
技术主管
私信 提问
windows 安装 storm 及 eclipse 调试 TopN 实例

一:安装JDK 下载地址:地址一 地址二 配置Java环境变量 JAVAHOME、Path、CLASSPATH三个值分别为(按照自己安装状况设置,此处供参考): D:javajdk1.8 %JAVAHOME%/bin;%JAVAHOME%/jre/bin ....

大数据之路
2012/06/08
535
1
Apache Storm 0.9.6/0.10.0 发布

Apache Storm 0.10.0 发布,此版本是个稳定版本,相比之前的 Beta 版本主要包括 bug 修复和改进: STORM-1108: Fix NPE in simulated time STORM-1106: Netty should not limit attempts to...

oschina
2015/11/06
4.7K
8
storm启动卡着不动,求高手

刚把storm集群部好,结果启动时卡着不动。 执行:./storm nimbus 提示: [root@WY_CentOS_100G bin]# ./storm nimbus Running: java -server -Dstorm.options= -Dstorm.home=/usr/local/src......

lja
2013/07/11
2.2K
1
Apache Slider + Storm

Apache Slider + Storm 系统环境 安装如下组件,部署可用环境 JDK 1.7.0_79 Apache Zookeeper 3.4.* Apache Zookeeper Apache Hadoop 2.6.* Apache Hadoop Apache Storm 0.9.4 Apache Storm......

Yulong_
2016/09/21
451
0
Storm实践3【TopN封装版】 - 1: 简要介绍

阅读前提:在阅读之前务必请先参看本ID的 Storm实践系列2 阅读目的 : 优化之前的封装 最新版本的Storm TOPN 整体的构造如下: 依然包含如下的几个package: 1 Spout: 模拟发射数据 2 bolt :...

止静
2014/09/13
296
1

没有更多内容

加载失败,请刷新页面

加载更多

最简单的获取相机拍照的图片

  import android.content.Intent;import android.graphics.Bitmap;import android.os.Bundle;import android.os.Environment;import android.provider.MediaStore;import andr......

MrLins
8分钟前
0
0
说好不哭!数据可视化深度干货,前端开发下一个涨薪点在这里~

随着互联网在各行各业的影响不断深入,数据规模越来越大,各企业也越来越重视数据的价值。作为一家专业的数据智能公司,个推从消息推送服务起家,经过多年的持续耕耘,积累沉淀了海量数据,在...

个推
10分钟前
2
0
第三方支付-返回与回调注意事项

不管是支付宝,微信,还是其它第三方支付,第四方支付,支付机构服务商只要涉及到钱的交易都要进行如下校验,全部成功了才视为成功订单 1.http请求是否成功 2.校验商户号 3.校验订单号及状态...

Shingfi
13分钟前
1
0
简述Java内存分配和回收策略以及Minor GC 和 Major GC(Full GC)

内存分配: 1. 栈区:栈可分为Java虚拟机和本地方法栈 2. 堆区:堆被所有线程共享,在虚拟机启动时创建,是唯一的目的是存放对象实例,是gc的主要区域。通常可分为两个区块年轻代和年老代。更...

DustinChan
18分钟前
3
0
Excel插入批注:可在批注插入文字、形状、图片

1.批注一直显示:审阅选项卡-------->勾选显示批注选项: 2.插入批注快捷键:Shift+F2 组合键 3.在批注中插入图片:鼠标右键点击批注框的小圆点【重点不可以在批注文本框内点击】----->调出批...

东方墨天
42分钟前
6
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部