标签：必须 无法 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("消息消费者已启动"); } }
运行结果：
消费者
生产者
标签：必须 无法 charset tac 错误 max 头部 binary current
原文地址：https://www.cnblogs.com/zhaoshizi/p/11111809.html
