Jstorm与RocketMQ整合

如果是经常关注阿里巴巴的朋友们,看到我这篇博客的题目,就知道我在参加今年的中间件比赛。

好了,废话不说,开始了。

首先我们知道,rocketmq的consumer有两种,一种是DefaultMQPushConsumer另外一个是DefaultMQPullConsumer

两个有什么区别呢?

对我们自己写的代码来说,使用push就是被动接受mq的消息,而使用pull就是需要主动的去mq上拉取消息。

那么再与jstorm集成的时候,选择哪个呢?

我最开始选择的是pull,后来遇到各种问题,放弃了。

选择push才是真爱呀。

具体怎么做呢?

在spot的open里初始化DefaultMQPushConsumer,registerMessageListener的时候填入自己,当然我们的spot实现了MessageListenerConcurrently,

然后在spot里面的consumeMessage里面写自己的逻辑,合适的时候,用collector发射消息就是了。

能上点干货么?

public class EmitPaymentSpot extends BaseRichSpout
    implements MessageListenerConcurrently{
	private static final long serialVersionUID = -3085994102089532269L;
	private SpoutOutputCollector collector;
	private transient DefaultMQPushConsumer consumer;

	@SuppressWarnings("rawtypes")
	public void open(Map conf, TopologyContext context,	SpoutOutputCollector collector) {
	     log.error("init DefaultMQPushConsumer");
		  consumer = new DefaultMQPushConsumer(RaceConfig.MetaConsumerGroup);

	     //   consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
		  consumer.setNamesrvAddr("ip:port")

	        try {
	        	consumer.subscribe(RaceConfig.MqTmallTradeTopic, "*");
			consumer.subscribe(RaceConfig.MqTaobaoTradeTopic, "*");
			consumer.subscribe(RaceConfig.MqPayTopic, "*");
		} catch (MQClientException e) {
			e.printStackTrace();
		}
	        consumer.registerMessageListener(this);
	        try {
			consumer.start();
		} catch (MQClientException e) {
			e.printStackTrace();
		}

	        log.error("Consumer Started.");
	        this.collector = collector;
	    }

	@Override
	public void nextTuple() {
		//do nothing
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		//...
	}

	@Override
	public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,ConsumeConcurrentlyContext context) {

		for (MessageExt msg : msgs) {
			byte[] body = msg.getBody();
			if (body.length == 2 && body[0] == 0 && body[1] == 0) {

				log.error("Got the end signal");
				collector.emit("stop",new Values("stop"));
				continue;
			}
			if (msg.getTopic().equals(RaceConfig.MqPayTopic)) {
				return doPayTopic(body);
			}else if (msg.getTopic().equals(RaceConfig.MqTaobaoTradeTopic)) {
				putTaobaoTradeToTair(body);
				return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
			} else if (msg.getTopic().equals(RaceConfig.MqTmallTradeTopic)) {
				putTmallTradeToTair(body);
				return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
			}else {
				return ConsumeConcurrentlyStatus.RECONSUME_LATER;
			}

		}
		return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
	}
}

当然还有第二种方式,是官方推荐的。

其实我觉得和我的方式差不多..

就是在生成consumer的时候使用工厂模式而已。

相关的代码,比较麻烦大家见

https://github.com/alibaba/jstorm/blob/master/jstorm-utility/jstorm-rocket-mq/src/main/java/com/alibaba/aloha/meta/MetaSpout.java

秒客网

Jstorm与RocketMQ整合

相关文章