首页 > 其他 > 详细

RocketMQ(三)——HelloWorld

时间:2020-01-15 18:58:10      阅读:130      评论:0      收藏:0      [点我收藏+]
代码示例
代码阐释
运行效果
内容补充
 

讲个9·3阅兵时程序员间流传的笑话:

  同学们,现在向我们走来的是程序员方阵!他们穿着拖鞋,披着毛巾,左手拿着键盘,右手举着鼠标,腋下夹着USB转换器。他们因睡眠不足而显得精神不振,喊着微弱的口号走过主席台,主席问候:程序员们辛苦了!程序员方队异口同声地答道:Hello World!

——研究一项技术,如果不提及“Hello World”,那指定是外行。
上篇博客,搭了一个最简单的集群——双主群集,这篇博客就利用这个环境,写一个简单的生产者、消费者,来快速体验一下RocketMQ的HelloWorld。
 

代码示例

Maven配置
引一下jar包,这里还是用3.2.6这一比较经典的版本

<dependency>
    <groupId>com.alibaba.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>3.2.6</version>
</dependency>

生产者
写一个简单的Producer类,来发送消息:

/**
 * Producer,发送消息
 */
public class Producer {
    public static void main(String[] args) throws MQClientException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("group_name");
        producer.setNamesrvAddr("192.168.2.222:9876;192.168.2.223:9876");
        producer.start();

        for (int i = 0; i < 100; i++) {
            try {
                Message msg = new Message("TopicTest",              // topic
                        "TagA",                                     // tag
                        ("HelloWorld - RocketMQ" + i).getBytes()    // body
                );
                SendResult sendResult = producer.send(msg);
                System.out.println(sendResult);
            } catch (Exception e) {
                e.printStackTrace();
                Thread.sleep(1000);
            }
        }

        producer.shutdown();
    }
}

消费者
写一个简单的Consumer类,来接收消息:

/**
 * Consumer,订阅消息
 */
public class Consumer {

    public static void main(String[] args) throws InterruptedException, MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group_name");
        consumer.setNamesrvAddr("192.168.2.222:9876;192.168.2.223:9876");
        /**
         * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
         * 如果非第一次启动,那么按照上次消费的位置继续消费
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe("TopicTest", "*");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg: msgs) {
                    try {
                        String topic = msg.getTopic();
                        String tags = msg.getTags();
                        String msgBody = new String(msg.getBody(),"utf-8");
                        System.out.println("收到消息--" + " topic:" + topic + " ,tags:" + tags + " ,msg:" +msgBody);
                    } catch (UnsupportedEncodingException e) {
                        e.printStackTrace();
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
        System.out.println("Consumer Started.");
    }
}
代码阐释
根据上面的生产者和消费者,说明几点内容:

GroupName:
无论生产者、消费者都必须给出GroupName,而且具有唯一性!
Topic、Tag:
生产到哪个Topic的哪个Tag下,消费者也是从Topic的哪个Tag进行消费,可见这个Tag有点类似于JMS Selector机制,即实现消息的过滤。
NameServer:
生产者、消费者需要设置NameServer地址。
消费方式:这里,采用的是Consumer Push的方式,即设置Listener机制回调,相当于开启了一个线程。
 
 

运行效果
光说不练嘴把式,来看一下运行效果:

生产者
技术分享图片

 

 仔细看看生产者结果输出,就会发现,有的消息发往broker-a,有的在broker-b上,自动实现了消息的负载均衡!

 

  • 消费者

技术分享图片

 

 这里消费消息是没有什么顺序的,以后我们在来谈消息的顺序性。

 

  • 管控台

 技术分享图片

 

 会发现消息分布在2个broker上。

技术分享图片

 

 

内容补充

启动顺序
务必保证先启动消费者进行Topic订阅,然后在启动生产者进行生产(否则极有可能导致消息的重复消费,重复消费,重复消费!重要的事情说三遍!)。而且在实际开发中,有时候不会批量的处理消息,而是原子性的,单线程的去一条一条的处理消息,这样就是实时的在处理消息。(批量的处理海量的消息,可以考虑Kafka)
持久化
在ActiveMQ中,生产消息的时候会提供是否持久化的选择,但是对于RocketMQ而言,消息是一定会被持久化的!
宕机处理
在多Master模式中,如果某个Master进程挂了,显然这台broker将不可用,上面的消息也将无法消费,要知道开源版本的RocketMQ是没有提供切换程序,来自动恢复故障的,因此在实际开发中,我们一般提供一个监听程序,用于监控Master的状态。
单批次消息消费数量
上面的消费者采用的是Push Consumer的方式,那么监听的Listener中的消息List到底是多少条呢?虽然提供了API,如consumer.setConsumeMessageBatchMaxSize(10),实际上即使设置了批量的条数,但是注意了,是最大是10,并不意味着每次batch的都是10,只有在消息有挤压的情况下才有可能。而且Push Consumer的最佳实践方式就是一条条的消费,如果需要batch,可以使用Pull Consumer。来做个测试:
1)  改一下消费者代码
技术分享图片

 

 

单批次消息消费数量测试代码


2) 运行效果–先启动消费者,再启动生产者

技术分享图片

 

 

先启消费者后启生产者运行效果


3) 运行效果–先启动生产者(这样消息会有挤压),再启动消费者

技术分享图片

 

 我这儿整理了比较全面的JAVA相关的面试资料,

技术分享图片

需要领取面试资料的同学,请加群:473984645

技术分享图片

 

 

 

RocketMQ(三)——HelloWorld

原文:https://www.cnblogs.com/1013wang/p/12198067.html

(0)
(0)
   
举报
评论 一句话评论(0
关于我们 - 联系我们 - 留言反馈 - 联系我们:wmxa8@hotmail.com
© 2014 bubuko.com 版权所有
打开技术之扣,分享程序人生!