RocketMQ在windows上安装和开发使用
2018-08-28 20:58
549 查看
一、下载rocketmq
http://rocketmq.apache.org/release_notes/release-notes-4.3.0/
二、启动name server
3.启动 broker server
4.生产者
5.消费者
设置环境变量
或者:
producer或消费者在使用之前需先设置nameserver的地址才能和RocketMQ通信,即:producer.setNamesrvAddr("服务器IP:9876")或者consumer.setNamesrvAddr("服务器IP:9876");
http://rocketmq.apache.org/release_notes/release-notes-4.3.0/
二、启动name server
> nohup sh bin/mqnamesrv & > tail -f ~/logs/rocketmqlogs/namesrv.log The Name Server boot success...
3.启动 broker server
> nohup sh bin/mqbroker -n localhost:9876 & > tail -f ~/logs/rocketmqlogs/broker.log The broker[%s, 172.30.30.233:10911] boot success...
4.生产者
public class OrderedProducer { public static void main(String[] args) throws Exception { //Instantiate with a producer group name. MQProducer producer = new DefaultMQProducer("example_group_name"); //Launch the instance. producer.start(); String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"}; for (int i = 0; i < 100; i++) { int orderId = i % 10; //Create a message instance, specifying topic, tag and message body. Message msg = new Message("TopicTestjjj", tags[i % tags.length], "KEY" + i, ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)); SendResult sendResult = producer.send(msg, new MessageQueueSelector() { @Override public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) { Integer id = (Integer) arg; int index = id % mqs.size(); return mqs.get(index); } }, orderId); System.out.printf("%s%n", sendResult); } //server shutdown producer.shutdown(); } }
5.消费者
public class OrderedConsumer { public static void main(String[] args) throws Exception { DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("example_group_name"); consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET); consumer.subscribe("TopicTest", "TagA || TagC || TagD"); consumer.registerMessageListener(new MessageListenerOrderly() { AtomicLong consumeTimes = new AtomicLong(0); @Override public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) { context.setAutoCommit(false); System.out.printf(Thread.currentThread().getName() + " Receive New Messages: " + msgs + "%n"); this.consumeTimes.incrementAndGet(); if ((this.consumeTimes.get() % 2) == 0) { return ConsumeOrderlyStatus.SUCCESS; } else if ((this.consumeTimes.get() % 3) == 0) { return ConsumeOrderlyStatus.ROLLBACK; } else if ((this.consumeTimes.get() % 4) == 0) { return ConsumeOrderlyStatus.COMMIT; } else if ((this.consumeTimes.get() % 5) == 0) { context.setSuspendCurrentQueueTimeMillis(3000); return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT; } return ConsumeOrderlyStatus.SUCCESS; } }); consumer.start(); System.out.printf("Consumer Started.%n"); } }
设置环境变量
export NAMESRV_ADDR=localhost:9876
或者:
producer或消费者在使用之前需先设置nameserver的地址才能和RocketMQ通信,即:producer.setNamesrvAddr("服务器IP:9876")或者consumer.setNamesrvAddr("服务器IP:9876");
相关文章推荐
- RocketMQ在windows上安装和开发使用
- RocketMQ在windows上安装和开发使用
- RocketMQ在windows上安装和eclipse开发使用
- RocketMQ在windows上安装和eclipse开发使用
- RocketMQ在windows上安装和开发使用
- RocketMQ在windows上安装和开发使用
- RocketMQ在windows上安装和开发使用
- RocketMQ在windows上安装和eclipse开发使用
- RocketMQ在windows上安装和开发使用
- RocketMQ在windows上安装和开发使用
- RocketMQ在windows上安装和开发使用
- RocketMQ在windows上安装和开发使用
- 使用VM安装windows系统的做开发系统的二三事
- Visual Studio 2008 和 Windows Driver Kits 驱动开发安装配置详解 - [软件安装使用技巧]
- windows下安装使用svn管理网站开发项目
- 快速构建Windows 8风格应用1-开发工具安装及模拟器使用
- 破解版windows 7(旗舰版)下安装并使用vagrant统一开发环境
- Win 8 系统下 JBoss AS 7 Final 的下载配置与使用——Windows J2EE开发安装序列之五
- QT 5 使用 BlueSoleil 开发WINDOWS蓝牙功能 ---------- 驱动安装
- 快速构建Windows 8风格应用1-开发工具安装及模拟器使用 推荐