RocketMQ搭建及demo
RocketMQ搭建及demo
六楼的宣言 发表于11个月前
RocketMQ搭建及demo
  • 发表于 11个月前
  • 阅读 47
  • 收藏 4
  • 点赞 0
  • 评论 0

新睿云服务器60天免费使用,快来体验!>>>   

一、RocketMQ机器硬件要求内存最好不要低于8G, 系统linux,且已经安装好JDK

二、安装文件下载地址:
http://mirror.bit.edu.cn/apache/incubator/rocketmq/4.0.0-incubating/rocketmq-all-4.0.0-incubating-bin-release.zip

三、下载RocketMQ安装文件并上传到服务器上后

解压 

unzip rocketmq-all-4.0.0-incubating-bin-release.zip

进入到解压目录下的bin目录中

启动 NameServer:

nohup sh mqnamesrv &

启动 broker

nohup sh mqbroker -n localhost:9876 &

使用jps命令可以看到有以下两个java程序运行中

为了便于其它机器调试访问,可临时将防火墙关闭:

service firewalld stop

 

四、Java 程序demo

maven依赖包:

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.0.0-incubating</version>
</dependency>

消息生产者demo代码:

package com.classtest.rocketmq;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

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

            Message msg = new Message("PushTopic",
                    "push",
                    "1",
                    "Just for test1.".getBytes());

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

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

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

            msg = new Message("PushTopic",
                    "push",
                    "1",
                    "Just for test3.".getBytes());

            result = producer.send(msg);
            System.out.println("id:" + result.getMsgId() +
                    " result:" + result.getSendStatus());
        } catch (Exception e) {
            e.printStackTrace();
        }finally{
            producer.shutdown();
        }
    }
}

消息消费者Java代码demo:

package com.classtest.rocketmq;

import java.util.List;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

public class Consumer {
    public static void main(String[] args){
        DefaultMQPushConsumer consumer =
                new DefaultMQPushConsumer("PushConsumer");
        consumer.setNamesrvAddr("192.168.133.141: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());
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        }
                    }
            );
            consumer.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

可以先运行消费者demo, 然后运行生产者demo时可以消费者demo的运行窗口输出消息

 

参考:

https://my.oschina.net/jayronwang/blog/861396

http://rocketmq.apache.org/docs/quick-start/

 

标签: CentOS RocketMQ Java
  • 打赏
  • 点赞
  • 收藏
  • 分享
共有 人打赏支持
六楼的宣言
粉丝 6
博文 21
码字总数 4289
×
六楼的宣言
如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
* 金额(元)
¥1 ¥5 ¥10 ¥20 其他金额
打赏人
留言
* 支付类型
微信扫码支付
打赏金额:
已支付成功
打赏金额: