RocketMQ的下载与环境配置

下载RocketMQ压缩包

点击下载

下拉页面,往下找

点击下载

解压到自定义目录,注意目录中不要有空格

配置RocketMQ的运行环境

配置环境变量

JAVA_HOME

注意JDK不要安装到有空格的目录,如Program File
1.在系统变量中新建JAVA_HOME

2.在系统变量的Path中添加jdk的bin目录

RocketMQ_HOME

1.在系统变量中新建JAVA_HOME

2.在系统变量的Path中添加RocketMQ解压目录下的bin目录

在解压目录下的conf下的broker.conf文件中添加本机外网地址

防止后续代码运行出现 MQClientException: No route info of this topic, TopicTest错误
端口号一定是9876,不要自定义(教训)

brokerIP1=10.68.151.248
namesrvAddr=10.68.151.248:9876

运行RocketMQ

先运行NameServer

在cmd中输入

start mqnamesrv -n 10.68.151.248:9876

启动nameserver,出现下面内容即为启动成功。

(NameServer是一个Broker与Topic路由的注册中心，支持Broker的动态注册与发现。)

再运行Broker

在RocketMQ的解压目录下打开cmd,执行下列命令

start mqbroker -n 10.68.151.248:9876 -c \broker.conf

出现如下内容,启动成功!

java代码发送消息和接收实例应用

添加RocketMQ的maven依赖

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

创建生产者类(发送消息)

public class RocketMqProducer {public static void main(String[] args) throws MQBrokerException, RemotingException, UnsupportedEncodingException, InterruptedException, MQClientException {new RocketMqProducer().defaultMQProducer();}public void defaultMQProducer() throws MQClientException, MQBrokerException, RemotingException, InterruptedException, UnsupportedEncodingException {//生产者,可以指定producer集群DefaultMQProducer producer = new DefaultMQProducer("producer_group_name");//设置name server的地址producer.setNamesrvAddr("10.68.151.248:9876");producer.start();System.out.println(producer.getNamesrvAddr());System.out.println(producer.getClientIP());System.out.println("启动了生产者producer");//message必须指定topic,和消息体body// 可以选择指定tag,key来进行细分messageMessage msgA = new Message("topicA", "这是topicA的消息,没有指定tag和key".getBytes(RemotingHelper.DEFAULT_CHARSET));Message msgB = new Message("topicB", "这是topicB的消息,没有指定tag和key".getBytes(RemotingHelper.DEFAULT_CHARSET));Message msgC = new Message("topicC","tag-a","这是topicC的消息,指定了tag-a".getBytes(RemotingHelper.DEFAULT_CHARSET));Message msgD = new Message("topicC","tag-b","这是topicC的消息,指定了tag-b".getBytes(RemotingHelper.DEFAULT_CHARSET));Message msgE = new Message("topicC","tag-a","key1","这是topicC的消息,指定了tag-a和key1".getBytes(RemotingHelper.DEFAULT_CHARSET));Message[] messages =new Message[]{msgA,msgB,msgC,msgD,msgE};//发送消息for (Message message : messages) {SendResult result = producer.send(message);System.out.println("消息发送成功:id:" + result.getMsgId() +" result:" + result.getSendStatus());}}
}

创建消费者类(接收消息)

public class RocketMqConsumer {public static void main(String[] args) throws MQClientException {new RocketMqConsumer().defaultMQPushConsumer();}public void defaultMQPushConsumer() throws MQClientException {//定义消费者,可以指定消费集群DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group_name");//同样的,指定name server 的地址consumer.setNamesrvAddr("10.68.151.248:9876");/*//订阅topicA下的所有消息consumer.subscribe("topicA","*");//一个consumer可以订阅多个topicconsumer.subscribe("topicB","*");*/consumer.subscribe("topicC","tag-a");//程序第一次启动从消息队列头取数据consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);//注册订阅消息(消息监听)consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,ConsumeConcurrentlyContext Context) {MessageExt msg = list.get(0);try {System.out.println("-收到消息:id-"+msg.getMsgId()+","+ new String(msg.getBody(), "UTF-8")+","+"keys: "+msg.getKeys());//System.out.println("msg全部信息:"+ msg.toString());} catch (UnsupportedEncodingException e) {e.printStackTrace();}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.out.println("consumer消费者启动");}
}