RocketMQ 可视化环境搭建和基础代码使用 (2)

解决方案：这是因为没有安装 Maven 或者没有配置 Maven 的环境变量导致的，下载 Maven 安装包，增加环境变量 MAVEN_HOME=maven安装目录  ，给 path 中添加 %MAVEN_HOME%\bin ，重新启动命令行工具（CMD）重新执行命令。

问题二：编译超慢的问题 解决方案：这是因为使用 Maven 数据源为国外源的问题导致的，只需要配置阿里的 Maven 源即可。
打开 Maven 目录下的 conf/setting.xml 给 mirrors 节点下添加如下内容：

<mirror> <id>alimaven</id> <name>aliyun maven</name> <url></url> <mirrorOf>central</mirrorOf> </mirror>

（4）运行插件

编译成功之后，进入 target 文件夹，执行 java -jar rocketmq-console-ng-1.0.1.jar 启动程序。
启动成功之后，在浏览器输入地址 :8066 进行访问，效果如下图：

3 基础使用

（1）添加引用 jar 包

pom.xml 添加以下代码：

<!-- https://mvnrepository.com/artifact/com.alibaba.rocketmq/rocketmq-client --> <dependency> <groupId>com.alibaba.rocketmq</groupId> <artifactId>rocketmq-client</artifactId> <version>3.6.2.Final</version> </dependency>

（2）添加生产者和消费者代码 public class RocketMQDemo { static final String MQ_NAMESRVADDR = "localhost:9876"; public static void main(String[] args) { // 分组名 String groupName = "myGroup-1"; // 主题名 String topicName = "myTopic-1"; // 标签名 String tagName = "myTag-1"; new Thread(() -> { try { producer(groupName, topicName, tagName); } catch (InterruptedException e) { e.printStackTrace(); } catch (RemotingException e) { e.printStackTrace(); } catch (MQClientException e) { e.printStackTrace(); } catch (MQBrokerException e) { e.printStackTrace(); } }).start(); new Thread(() -> { try { consumer(groupName, topicName, tagName); } catch (MQClientException e) { e.printStackTrace(); } }).start(); } /** * @Description 生产者 * @Author wanglei * @Param [groupName 分组名, topicName 主题名, tagName 标签名] **/ public static void producer(String groupName, String topicName, String tagName) throws InterruptedException, RemotingException, MQClientException, MQBrokerException { DefaultMQProducer producer = new DefaultMQProducer(groupName); producer.setNamesrvAddr(MQ_NAMESRVADDR); producer.start(); String body = "Hello, 老王"; Message message = new Message(topicName, tagName, body.getBytes()); producer.send(message); producer.shutdown(); } /** * @Description 消费者 * @Author wanglei * @Param [groupName 分组名, topicName 主题名, tagName 标签名] **/ public static void consumer(String groupName, String topicName, String tagName) throws MQClientException { DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName); consumer.setNamesrvAddr(MQ_NAMESRVADDR); consumer.subscribe(topicName, tagName); consumer.registerMessageListener(new MessageListenerConcurrently() { @Override public ConsumeConcurrentlyStatus consumeMessage( List<MessageExt> msgs, ConsumeConcurrentlyContext context) { for (MessageExt msg : msgs) { System.out.println(new String(msg.getBody())); } return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); consumer.start(); } }

以上程序执行结果如下：