rocketMq-demo

• 前言
• jar包预备
• demo

前言

前面一面博文写了RocketMq的配置安装，这里记录一下自己的demo。

jar包预备

我用的maven来管理项目，加入maven依赖：
com.alibaba.rocketmq rocketmq-client 3.2.6

Producer

package com.yyp.test;

/**
 * Created by yyp on 15-12-18.
 */

import com.alibaba.rocketmq.common.message.Message;

import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;

public class Producer {
public static void main(String[] args) {
DefaultMQProducer producer = new DefaultMQProducer("Producer");
producer.setNamesrvAddr("192.168.0.1:9876");
try {
producer.start();
int i = 0;

while (true) {
i++;
Message msg = new Message("PushTopic",
"push",
"1",
("Just for yyp" + i).getBytes());
SendResult result = producer.send(msg);

   /* msg = new Message("PushTopic",
"push",
"2",
"Just for test.".getBytes());

result = producer.send(msg);
System.out.println("id:" + result.getMsgId() +
" result:" + result.getSendStatus());*/

msg = new Message("PullTopic",
"pull",
"1",
"Just for test.".getBytes());
result = producer.send(msg);
System.out.println("id:" + result.getMsgId() +
" result:" + result.getSendStatus());
if (i % 10 == 0) {
Thread.sleep(20000);
}
}

} catch (Exception e) {
e.printStackTrace();
} finally {
// producer.shutdown();
}
}
}

Consumer

package com.yyp.test;

/**
 * Created by yyp on 15-12-18.
 */

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.common.message.MessageExt;
import java.util.List;

public class Consumer {
public static void main(String[] args) {
DefaultMQPushConsumer consumer =
new DefaultMQPushConsumer("PushConsumer");
consumer.setNamesrvAddr("192.168.0.1:9876");
try {
//订阅PushTopic下Tag为push的消息
consumer.subscribe("PushTopic", "push");
//程序第一次启动从消息队列头取数据
consumer.setConsumeFromWhere(
ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.registerMessageListener(
new MessageListenerConcurrently() {
public ConsumeConcurrentlyStatus consumeMessage(
List<MessageExt> list,
ConsumeConcurrentlyContext Context) {
   /* Message msg = list.get(0);
System.out.println(msg.toString());*/
MessageExt messageExt=list.get(0);
System.out.println("msg:\t"+new String(messageExt.getBody()));
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
}
);
consumer.start();
} catch (Exception e) {
e.printStackTrace();
}
}
}