RocketMq 实际案例–普通消息的发送 |我的资源库露水湾

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<MessageExt> 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