RocketMq 实际案例–普通消息的发送

@(消息中间件)[RocketMq 实例]

学习 rocketMq 最根本的是要先学会用嘛，在创建 rocketMq 的第一个案例的时候，碰到很多坑，可以记录下。

第一步：启动 namesrv 和 broker

官方文档未告知的坑:

问题 1： 如果在单机环境下，没有 Docker 的部署环境，rocketMq 是会默认走 VIPChannel，这时候会在我们发送消息的时候报

rocketMqb 报错.png

解决方法： 修改 bin／conf 中的 broker.conf 文件，指定 namesrvAdd 和 brokerIp

问题二： 在实际发送过程中，会出现无法自动创建不存在 topic 错误

解决方法: nohup sh mqbroker -n localhost:9876 autoCreateTopicEnable=true & 启动 broker 的时候要加上 autoCreateTopicEnable=true

第二步：启动 rocketMq 的控制台

对于启动控制台，是方便我们的查看。控制台的原码在 github 中是一个扩展项目https://github.com/apache/rocketmq-externals

项目本身是 spring-boot 的实现，所以之间启动 APP.java 即可(后面需要学习 spring-boot 和 rocketMq 的整合)

第三步：编写我们的生产者和消费者

Producer (组)

package com.zzjmay.rocketMq.producer;

import com.alibaba.fastjson.JSON;

import org.apache.rocketmq.client.producer.DefaultMQProducer;

import org.apache.rocketmq.client.producer.SendResult;

import org.apache.rocketmq.common.message.Message;

/**

* 生产者

* Created by zzjmay on 2018/1/14.

*/

public class RocketProducer2 {

public static void main(String[] args) {

//1.创建一个生产者,需要指定 Producer 的分组，

DefaultMQProducer defaultMQProducer = new DefaultMQProducer("Jony-P");

//2.设置命名服务的地址,默认是去读取 conf 文件下的配置文件 rocketmq.namesrv.addr

defaultMQProducer.setNamesrvAddr("127.0.0.1:9876");

try{

//3.启动生产者

defaultMQProducer.start();

for(int i=0;i<10;i++) {

String text = "你好呀，ALIBABA----"+i;

//4.最基本的生产模式 topic+文本信息

Message msg = new Message("topic_orderCreate", "Tag-B", text.getBytes());

//5.获取发送响应

SendResult sendResult = defaultMQProducer.send(msg);

System.out.println("###发送结果 result:" + JSON.toJSONString(sendResult));

}

}catch (Exception e){

e.printStackTrace();

}finally {

//6.释放生产者

defaultMQProducer.shutdown();

}

}

}

Consumer (组)

package com.zzjmay.rocketMq.consumer;

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.client.exception.MQClientException;

import org.apache.rocketmq.common.consumer.ConsumeFromWhere;

import org.apache.rocketmq.common.message.MessageExt;

import java.util.Date;

import java.util.List;

/**

* 消费者

* Created by zzjmay on 2018/1/14.

*/

public class RocketConsumer {

public static void main(String[] args) {

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("Consumer-g");

consumer.setNamesrvAddr("127.0.0.1:9876");

consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

try {

//第二个参数表示消费匹配的 tag * 表示 topic 所有的 tag

consumer.subscribe("topic_orderCreate","Tag-B");

//2. 注册消费者监听

consumer.registerMessageListener(new MessageListenerConcurrently() {

/**

* msgs 表示消息体

* @param msgs

* @param context

* @return

*/

public ConsumeConcurrentlyStatus consumeMessage(List msgs, ConsumeConcurrentlyContext context) {

for(MessageExt messageExt:msgs){

try {

System.out.println(new Date()+new String(messageExt.getBody(),"UTF-8"));

}catch (Exception e){

e.printStackTrace();

}

}

return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

}

});

//3.consumer 启动

consumer.start();

System.out.println("消费端起来了哈.........");

} catch (MQClientException e) {

e.printStackTrace();

}

}

}

最后查看效果在 console

监控图

集群信息

作者：zzjmay

链接：https://www.jianshu.com/p/94473cb87e0b

来源：简书

简书著作权归作者所有，任何形式的转载都请联系作者获得授权并注明出处。

露水湾 , 版权所有丨如未注明 , 均为原创丨本网站采用BY-NC-SA协议进行授权

转载请注明原文链接：RocketMq 实际案例–普通消息的发送