RocketMQ 入门使用详解[通俗易懂]

RocketMQ 入门使用详解[通俗易懂]RocketMQ是阿里巴巴在2012年开源的分布式消息中间件,目前已经捐赠给Apache基金会,已经于2016年11月成为 Apache 孵化项目,相信RocketMQ的未来会发挥着越来越大的作用,将有更多的开发者因此受益。 本文仅对RocketMQ的简单实用做入门性介绍,不对RocketMQ的底层原理进行深入介绍,后续文章将对RocketMQ的原理做详细介绍。

大家好,又见面了,我是你们的朋友全栈君。

       RocketMQ是阿里巴巴在2012年开源的分布式消息中间件,目前已经捐赠给Apache基金会,已经于2016年11月成为 Apache 孵化项目,相信RocketMQ的未来会发挥着越来越大的作用,将有更多的开发者因此受益。

        本文仅对RocketMQ的简单实用做入门性介绍,不对RocketMQ的底层原理进行深入介绍,后续文章将对RocketMQ的原理做详细介绍。

        RocketMQ的Maven依赖:

   

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

     MQ的消费类RocketMQConsumer.java:

    

package com.lance.rocketMQ.RocketMQ;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.MessageListener;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;

import java.util.UUID;


/**
 * Created by lance on 2017/2/10.
 */
public class RocketMQConsumer {

    private DefaultMQPushConsumer consumer;

    private MessageListener listener;

    protected String nameServer;

    protected String groupName;

    protected String topics;

    public RocketMQConsumer(MessageListener listener, String nameServer, String groupName, String topics) {
        this.listener = listener;
        this.nameServer = nameServer;
        this.groupName = groupName;
        this.topics = topics;
    }

    public void init() {
        consumer = new DefaultMQPushConsumer(groupName);
        consumer.setNamesrvAddr(nameServer);
        try {
            consumer.subscribe(topics, "*");
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        consumer.setInstanceName(UUID.randomUUID().toString());
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.registerMessageListener((MessageListenerConcurrently) this.listener);

        try {
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        System.out.println("RocketMQConsumer Started! group=" + consumer.getConsumerGroup() + " instance=" + consumer.getInstanceName()
        );
    }


}

       MQ消息的监听接口类RocketMQListener.java

      

        

package com.lance.rocketMQ.RocketMQ;

import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * Created by lance on 2017/2/10.
 */
public class RocketMQListener  implements MessageListenerConcurrently {


    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
//        System.out.println("get data from rocketMQ:" + msgs);
        for (MessageExt message : msgs) {

            String msg = new String(message.getBody());
            System.out.println("msg data from rocketMQ:" + msg);
        }

        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}

        MQ消息的生产者类RocketMQProducer.java

package com.lance.rocketMQ.RocketMQ;

import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.client.producer.SendStatus;
import com.alibaba.rocketmq.common.message.Message;

import java.util.UUID;

/**
 * Created by lance on 2017/2/10.
 */
public class RocketMQProducer {

    private DefaultMQProducer sender;

    protected String nameServer;

    protected String groupName;

    protected String topics;

    public void init() {
        sender = new DefaultMQProducer(groupName);
        sender.setNamesrvAddr(nameServer);
        sender.setInstanceName(UUID.randomUUID().toString());
        try {
            sender.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    public RocketMQProducer(String nameServer, String groupName, String topics) {
        this.nameServer = nameServer;
        this.groupName = groupName;
        this.topics = topics;
    }

    public void send(Message message) {

        message.setTopic(topics);

        try {
            SendResult result = sender.send(message);
            SendStatus status = result.getSendStatus();
            System.out.println("messageId=" + result.getMsgId() + ", status=" + status);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

       测试RocketMQ的消费 RocketMQConsumerTest.java

package com.lance.rocketMQ.RocketMQ;

/**
 * Created by lance on 2017/2/10.
 */
public class RocketMQConsumerTest {


    public static void main(String[] args) {


        String mqNameServer = "172.10.254.2:9876";
        String mqTopics = "MQ-MSG-TOPICS-TEST";

        String consumerMqGroupName = "CONSUMER-MQ-GROUP";
        RocketMQListener mqListener = new RocketMQListener();
        RocketMQConsumer mqConsumer = new RocketMQConsumer(mqListener, mqNameServer, consumerMqGroupName, mqTopics);
        mqConsumer.init();


        try {
            Thread.sleep(1000 * 60L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }
}

          run RocketMQConsumerTest.java 之后,控制台输出:

       

RocketMQConsumer Started! group=CONSUMER-MQ-GROUP instance=1eb7d308-4414-4658-90b5-e2cae3b793eb

            结果分析: 此时MQ对应的TOPIC中并没有响应的消息,故收不到消息,仅看到MQ消费者正常启动信息。

       

        MQ的生产者测试类:RocketMQProducerTest.java

       

package com.lance.rocketMQ.RocketMQ;

import com.alibaba.rocketmq.common.message.Message;

/**
 * Created by lance on 2017/2/10.
 */
public class RocketMQProducerTest {

    public static void main(String[] args) {

        String mqNameServer = "172.10.254.2:9876";
        String mqTopics = "MQ-MSG-TOPICS-TEST";

        String producerMqGroupName = "PRODUCER-MQ-GROUP";
        RocketMQProducer mqProducer = new RocketMQProducer(mqNameServer, producerMqGroupName, mqTopics);
        mqProducer.init();


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

            Message message = new Message();
            message.setBody(("I send message to RocketMQ " + i).getBytes());
            mqProducer.send(message);
        }



    }

}

          run RocketMQProducerTest.java 之后,RocketMQProducerTest.java 对应的控制台输出为:

                

messageId=0A71290100002A9F00000003D0BB0832, status=SEND_OK
messageId=0A71290100002A9F00000003D0BB08BB, status=SEND_OK
messageId=0A71290100002A9F00000003D0BB0944, status=SEND_OK
messageId=0A71290100002A9F00000003D0BB09CD, status=SEND_OK
messageId=0A71290300002A9F000000005440AEED, status=SEND_OK

         结果分析:表明所有消息都已经正常发送,且被RocketMQ正常接收。

          此时查看RocketMQConsumerTest.java对应的控制台输出发生改变,输出内容变更如下:

         

RocketMQConsumer Started! group=CONSUMER-MQ-GROUP instance=1eb7d308-4414-4658-90b5-e2cae3b793eb
msg data from rocketMQ:I send message to RocketMQ 1
msg data from rocketMQ:I send message to RocketMQ 0
msg data from rocketMQ:I send message to RocketMQ 3
msg data from rocketMQ:I send message to RocketMQ 2
msg data from rocketMQ:I send message to RocketMQ 4

看,简单吧!

备注:小编自己使用了Apache版本的RocketMQ(即RocketMQ 4.*),发现只需要更改import的package的路径而已,不需要修改其他代码,请参考。

RocketMQ的重复问题解决方式:
a.MQ的消费端执行的操作具有幂等性,即无论多少次重复执行,其结果是一样的;
b.MQ的消费端做重复校验,比如将受到MQ消息的唯一编号保存到Redis中,即每次收到消息时,将检查唯一编号是否已经在Redis中,如果存在说明消息重复;否则将唯一编号放入到Redis中,可以根据系统需要设置唯一编号在Redis中的过期时间,以防止Redis溢出。

 

参考链接:1.分布式开放消息系统(RocketMQ)的原理与实践 (强烈推荐)

                    2.RocketMQ捐赠给Apache那些鲜为人知的故事

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/149571.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)


相关推荐

  • 传统请求风格 VS RestFul 风格

    传统请求风格 VS RestFul 风格RestFul风格概念Restful就是一个资源定位及资源操作的风格。不是标准也不是协议,只是一种风格。基于这个风格设计的软件可以更简洁,更有层次,更易于实现缓存等机制。功能资源:互联网所有的事物都可以被抽象为资源资源操作:使用POST、DELETE、PUT、GET,使用不同方法对资源进行操作。分别对应添加、删除、修改、查询。传统方式操作资源:通过不同的参数来实现不同的效果!方法单一,post和get​ http://127.0.0.1/item/queryItem.actio

  • connectionstrings-sql server 2012[通俗易懂]

    connectionstrings-sql server 2012[通俗易懂]最近发现了超全connectionstrings网站:https://www.connectionstrings.comSQLServer2012connectionstrings.NETFrameworkDataProviderforSQLServerStandardSecurityServer=myServerAddress;Databas…

  • java中方法的定义

    java中方法的定义

  • java代码生成器,springboot代码生成器—增加更新,查询功能(持续更新)

    java代码生成器,springboot代码生成器—增加更新,查询功能(持续更新)时隔一周多,今天终于抽出时间来更新一波代码生成器,最近公司让我研究rpa,弄得焦头烂额的,话不多说,进入正题。之前有朋友让我讲一下代码生成器的原理,这篇博客就大体描述一下,以后慢慢细致讲解。双击codeMan.exe,众所周知,java做成exe程序很麻烦,在这里我是利用了.net的ikvm插件把jar包直接编译成了exe程序,这个启动界面会连接我的服务器,去检查版本更新,如果有更新就会在…

  • hive 数据类型_sv数据类型

    hive 数据类型_sv数据类型本篇来学习一下Hive中的数据类型,以及如何在HiveSQL中去处理这些数据类型。Hive的数据类型可以分为基本类型、字符串类型、日期与时间戳类型、集合类型,我们分别进行介绍:1、基…

  • VMware许可证过期_windows许可证过期什么原因

    VMware许可证过期_windows许可证过期什么原因激活密钥VMware2017v14.x永久许可证激活密钥FF31K-AHZD1-H8ETZ-8WWEZ-WUUVACV7T2-6WY5Q-48EWP-ZXY7X-QGUWD这表文章解决了大部分的常见问题,大家可以去看看http://www.zdfans.com/5928.html

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号