码迷,mamicode.com
首页 > 其他好文 > 详细

RocketMQ4.5.1环境搭建及示例

时间:2019-07-01 00:38:44      阅读:128      评论:0      收藏:0      [点我收藏+]

标签:必须   无法   charset   tac   错误   max   头部   binary   current   

一、Windows环境搭建RocketMQ

1. 下载RocketMQ Binary压缩包,并解压缩,我的安装目录为E:\programs\rocketmq\rocketmq-all-4.5.1

2. 配置环境变量:ROCKETMQ_HOME,其值为RocektMQ的安装目录

技术图片

3. 启动NameServer

在命令行中进入RocketMQ安装目录下的bin目录,执行start mqnamesrv.cmd,执行完成后会弹出一个新窗口,不要关闭该窗口

技术图片

4. 启动Broker

在命令行中进入RocketMQ安装目录下的bin目录,执行start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true,执行完成后会弹出一个新窗口,不要关闭该窗口

首次执行中遇到了如下问题:错误: 找不到或无法加载主类 Files\Java\jdk1.8.0_144\lib\dt.jar;C:\Program

看错误提示应该是CLASSPATH变量中有空格,导致读取失败

技术图片

可以通过修改bin目录下的runbroker.cmd文件,找到其中的倒数第二行: set "JAVA_OPT=%JAVA_OPT% -cp %CLASSPATH%",在%CLASSPATH%两边加双引号,引起来,这样就可以识别CLASSPATH变量中的空格。再次执行start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true,成功。

技术图片

技术图片

 

二、ubuntu环境搭建RocketMQ
1. 下载RocketMQ的源码,并解压缩

unzip rocketmq-all-4.5.1-source-release.zip

2. 在解压缩的目录中,使用maven构建项目,构建完成后,进入rocketmq目录

cd rocketmq-all-4.4.0/
mvn -Prelease-all -DskipTests clean install -U
cd distribution/target/apache-rocketmq

3. 视情况修改runbroker.sh和runserver.sh文件中JVM参数,默认参数设置的内存比较大,个人电脑运行不起来,需要调小

runbroker.sh中

原配置:JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g -Xmn4g"

修改为:JAVA_OPT="${JAVA_OPT} -server -Xms128m -Xmx256m -Xmn256m"

runserver.sh中

原配置:JAVA_OPT="${JAVA_OPT} -server -Xms4g -Xmx4g -Xmn2g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"

修改为:JAVA_OPT="${JAVA_OPT} -server -Xms128m -Xmx256m -Xmn256m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"

 

4. 启动nameserver:

在构建完成的rocketmq目录下,执行如下语句,启动nameserver

nohup sh bin/mqnamesrv &

查看日志

tail -f ~/logs/rocketmqlogs/namesrv.log

5. 启动broker

启动broker语句和查看日志语句

nohup sh bin/mqbroker -n localhost:9876 &
tail -f ~/logs/rocketmqlogs/broker.log 

6. 关闭broker和server语句

sh bin/mqshutdown broker

sh bin/mqshutdown namesrv

 

三、示例

 生产者

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

public class Producer {
    public static void main(String[] args) throws Exception {
        //创建一个消息生产者,并设置一个消息生产者组
        DefaultMQProducer producer = new DefaultMQProducer("zs_producer_group");
        //指定NameServer地址
        producer.setNamesrvAddr("localhost:9876");
        //初始化Producer,在整个应用生命周期中只需要初始化一次
        producer.start();
        for(int i = 0;i<100;i++) {
            //创建一个消息对象,指定其主题、标签和消息内容
            Message msg = new Message(
                    "topic_example_java"  /*消息主题名*/,
                    "TagA"    /*消息标签*/,
                    ("Hello Java demo RocketMQ" + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /*消息内容*/
                    );
            //发送消息并返回结果
            SendResult sendResult = producer.send(msg);
            
            System.out.printf("%s%n", sendResult);
        }
        //一旦生产者实例不再被使用,则将其关闭,包括清理资源、关闭网络连接等
        producer.shutdown();    
    }
}

消费者

import java.util.Date;
import java.util.List;
import java.io.UnsupportedEncodingException;

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.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

public class Consumer {
    public static void main(String[] args) throws Exception {
        //创建一个消息消费者,并设置一个消息消费者组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("zs_consumer_group");
        //指定NameServer地址
        consumer.setNamesrvAddr("localhost:9876");
        //设置Consumer第一次启动时是从队列头部还是队列尾部开始消费的
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //订阅指定Topic下的所有消息
        consumer.subscribe("topic_example_java","*");
        //注册消息监听器
        consumer.registerMessageListener((List<MessageExt> list, ConsumeConcurrentlyContext context) ->{
            //默认list里只有一条消息,可以通过设置参数来批量接收消息
            if(list!= null) {
                for(MessageExt ext: list) {
                    try {
                        System.out.println(new Date()+ new String(ext.getBody(),"UTF-8"));
                    }catch(UnsupportedEncodingException e) {
                        e.printStackTrace();
                    }
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        //消息者对象在使用之前必须要调用start方法初始化
        consumer.start();
        System.out.println("消息消费者已启动");
    }
}

运行结果:

消费者

技术图片

生产者

技术图片

 

RocketMQ4.5.1环境搭建及示例

标签:必须   无法   charset   tac   错误   max   头部   binary   current   

原文地址:https://www.cnblogs.com/zhaoshizi/p/11111809.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!