RocketMQ入门与架构解析

需积分: 0 8 浏览量更新于2024-07-06 收藏 10.09MB PDF 举报

"RocketMQ基础入门，包括RocketMQ的发展历程、使用消息中间件的原因以及RocketMQ的主要版本。" Apache RocketMQ 是一个高性能、分布式的消息中间件，最初由阿里巴巴开发，后来成为Apache顶级项目。它主要用于解决应用之间的解耦、流量削峰和数据分发问题。在大型分布式系统中，消息队列如RocketMQ扮演着关键角色，允许生产者无须直接与消费者交互，只需将消息发送到队列，而消费者可以从队列中获取并处理数据，这样降低了系统的耦合度，提高了开发效率。消息中间件的主要优点： 1. **应用解耦**：通过消息队列，系统间的依赖关系变得松散，生产者和消费者可以独立开发和更新，减少了因改动一方而导致的连锁反应。 2. **流量削峰**：在高并发情况下，消息队列能暂时存储大量请求，避免后端服务因瞬时流量过大而崩溃。 3. **数据分发**：数据可以在多个系统间自由流动，支持多消费者模式，方便数据的处理和分析。 RocketMQ的发展历程： 1. Metaq 1.x 版本是RocketMQ的前身，是对Kafka的Java重写。 2. Metaq 2.x 对存储进行了优化，解决了Kafka在大规模交易中的横向扩展问题。 3. RocketMQ 3.0，即从Metaq 2.x更名而来，是公开开源的起点。 4. 现在广泛使用的RocketMQ 4.x版本，自2017年起成为Apache顶级项目。 5. RocketMQ 5.x 正在内部推进，聚焦于CloudNative（云原生）方向。 RocketMQ在阿里巴巴及其生态中的应用广泛，比如Aliware MQ，适用于微服务、流计算、物联网(IoT)、异步解耦和数据同步等多种场景。此外，OpenMessaging项目由阿里巴巴发起，旨在创建一套统一的消息和流处理标准，并已融入RocketMQ和整个阿里云平台。 RocketMQ的物理架构主要包括以下组件： 1. **NameServer**：路由注册与发现中心，管理所有Topic和Broker的映射关系。 2. **Producer**：消息生产者，负责发送消息到消息队列。 3. **Consumer**：消息消费者，订阅并消费消息。 4. **Broker**：消息服务器，存储和转发消息。 5. **Message Queue**：消息的存储单元，消息在Queue中排队等待消费。 RocketMQ提供了多种消息模型，如Push模型和Pull模型，以及集群消费和广播消费等模式，以适应不同应用场景的需求。其高可用性和可扩展性使其成为企业级分布式系统中的首选消息中间件。

批量消息

批量发送消息能显著提高传递小消息的性能。限制是这些批量消息应该有相同的 topic，相同的

waitStoreMsgOK（集群时会细讲），而且不能是延时消息。此外，这一批消息的总大小不应超过

4MB。

代码演示

   consumer.start();

 }

}

// 生产者

public class BatchProducer {

 public static void main(String[] args) throws Exception {

   // 实例化消息生产者Producer

   DefaultMQProducer producer = new DefaultMQProducer("BatchProducer");

   // 设置NameServer的地址

   producer.setNamesrvAddr("localhost:9876");

   // 启动Producer实例

   producer.start();

   String topic = "BatchTest";

   List<Message> messages = new ArrayList<>();

   messages.add(new Message(topic, "Tag", "OrderID001", "Hello world

0".getBytes()));

   messages.add(new Message(topic, "Tag", "OrderID002", "Hello world

1".getBytes()));

   messages.add(new Message(topic, "Tag", "OrderID003", "Hello world

2".getBytes()));

   try {

     producer.send(messages);

   } catch (Exception e) {

     producer.shutdown();

     e.printStackTrace();

   }

   // 如果不再发送消息，关闭Producer实例。

   producer.shutdown();

 }

}

// **消费者**

public class BatchComuser {

 public static void main(String[] args) throws Exception {

   // 实例化消息生产者,指定组名

   DefaultMQPushConsumer consumer = new

DefaultMQPushConsumer("BatchComsuer");

   // 指定Namesrv地址信息.

   consumer.setNamesrvAddr("localhost:9876");

   // 订阅Topic

   consumer.subscribe("BatchTest", "*");

   //负载均衡模式消费

   consumer.setMessageModel(MessageModel.CLUSTERING);

   // 注册回调函数，处理消息

   consumer.registerMessageListener(new MessageListenerConcurrently() {

批量切分

如果消息的总长度可能大于 4MB 时，这时候最好把消息进行分割

代码演示

我们需要发送 10 万元素的数组，这个量很大，怎么快速发送完。同时每一次批量发送的消息大小

不能超过 4M 具体见代码

     @Override

     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt>

msgs,

       ConsumeConcurrentlyContext context) {

       System.out.printf("%s Receive New Messages: %s %n",

         Thread.currentThread().getName(), msgs);

       return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

     }

   });

   //启动消息者

   consumer.start();

   System.out.printf("Consumer Started.%n");

 }

}

* Licensed to the Apache Software Foundation (ASF) under one or more

* contributor license agreements. See the NOTICE file distributed with

* this work for additional information regarding copyright ownership.

* The ASF licenses this file to You under the Apache License, Version 2.0

* (the "License"); you may not use this file except in compliance with

* the License. You may obtain a copy of the License at

*   http://www.apache.org/licenses/LICENSE-2.0

* Unless required by applicable law or agreed to in writing, software

* distributed under the License is distributed on an "AS IS" BASIS,

* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

* See the License for the specific language governing permissions and

* limitations under the License.

package org.apache.rocketmq.example.batch;

import java.util.ArrayList;

import java.util.Iterator;

import java.util.List;

import java.util.Map;

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

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

/**

* @author 【享学课堂】 King老师

* 批量消息-超过4m-生产者

public class SplitBatchProducer {

 public static void main(String[] args) throws Exception {

   // 实例化消息生产者Producer

   DefaultMQProducer producer = new DefaultMQProducer("BatchProducer");

   // 设置NameServer的地址

   producer.setNamesrvAddr("localhost:9876");

   // 启动Producer实例

   producer.start();

   //large batch

   String topic = "BatchTest";

   List<Message> messages = new ArrayList<>(100 * 1000);

   //10万元素的数组

   for (int i = 0; i < 100 * 1000; i++) {

     messages.add(new Message(topic, "Tag", "OrderID" + i, ("Hello world

" + i).getBytes()));

   }

   //把大的消息分裂成若干个小的消息（1M左右）

   ListSplitter splitter = new ListSplitter(messages);

   while (splitter.hasNext()) {

     List<Message> listItem = splitter.next();

     producer.send(listItem);

     Thread.sleep(100);

   }

   // 如果不再发送消息，关闭Producer实例。

   producer.shutdown();

   System.out.printf("Consumer Started.%n");

 }

}

class ListSplitter implements Iterator<List<Message>> {

 private int sizeLimit = 1000 * 1000;//1M

 private final List<Message> messages;

 private int currIndex;

 public ListSplitter(List<Message> messages) {

   this.messages = messages;

 }

 @Override

 public boolean hasNext() {

   return currIndex < messages.size();

 }

 @Override

 public List<Message> next() {

   int nextIndex = currIndex;

   int totalSize = 0;

   for (; nextIndex < messages.size(); nextIndex++) {

     Message message = messages.get(nextIndex);

     int tmpSize = message.getTopic().length() +

message.getBody().length;

     Map<String, String> properties = message.getProperties();

     for (Map.Entry<String, String> entry : properties.entrySet()) {

       tmpSize += entry.getKey().length() + entry.getValue().length();

     }

     tmpSize = tmpSize + 20; // 增加日志的开销20字节

     if (tmpSize > sizeLimit) {

过滤消息

Tag 过滤

在大多数情况下，TAG 是一个简单而有用的设计，其可以来选择您想要的消息。

       //单个消息超过了最大的限制（1M）

       //忽略,否则会阻塞分裂的进程

       if (nextIndex - currIndex == 0) {

         //假如下一个子列表没有元素,则添加这个子列表然后退出循环,否则只是退出循环

         nextIndex++;

       }

       break;

     }

     if (tmpSize + totalSize > sizeLimit) {

       break;

     } else {

       totalSize += tmpSize;

     }

   }

   List<Message> subList = messages.subList(currIndex, nextIndex);

   currIndex = nextIndex;

   return subList;

 }

 @Override

 public void remove() {

   throw new UnsupportedOperationException("Not allowed to remove");

 }

}

// 核心思想就是自己去定义自己的迭代器, 这是一个很好的思路, 我们做切分也可以自己定义迭代器, 使用

方不用关心细节

// 生产

public class TagFilterProducer {

 public static void main(String[] args) throws Exception {

   DefaultMQProducer producer = new DefaultMQProducer("TagFilterProducer");

   //todo 指定Namesrv地址信息.

   producer.setNamesrvAddr("localhost:9876");

   producer.start();

   String[] tags = new String[]{"TagA", "TagB", "TagC"};

   for (int i = 0; i < 60; i++) {

     Message msg = new Message("TagFilterTest",

       tags[i % tags.length],

       "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));

     SendResult sendResult = producer.send(msg);

     System.out.printf("%s%n", sendResult);

   }

   producer.shutdown();

消费者将接收包含 TagA (区分大小写)或 TAGB 或 TAGC 的消息。但是限制是一个消息只能有一个标

签，这对于复杂的场景可能不起作用。在这种情况下，可以使用 SQL 表达式筛选消息。SQL

特性可以通过发送消息时的属性来进行计算。

 }

}

// 消费

public class TagFilterConsumer {

 public static void main(String[] args)

   throws InterruptedException, MQClientException, IOException {

   DefaultMQPushConsumer consumer = new

DefaultMQPushConsumer("TagFilterComsumer");

   //todo 指定Namesrv地址信息.

   consumer.setNamesrvAddr("localhost:9876");

   consumer.subscribe("TagFilterTest", "TagA || TAGB || TAGC"); // 区分大小写

   consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

   consumer.registerMessageListener(new MessageListenerConcurrently() {

     @Override

     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt>

msgs,

       ConsumeConcurrentlyContext context) {

       try {

         for (MessageExt msg : msgs) {

           String topic = msg.getTopic();

           String msgBody = new String(msg.getBody(), "utf-8");

           String msgPro = msg.getProperty("a");

           String tags = msg.getTags();

           System.out.println(

             "收到消息：" + " topic :" + topic + " ,tags : " + tags

+ " ,a : " + msgPro

               + " ,msg : " + msgBody);

         }

       } catch (Exception e) {

         e.printStackTrace();

         return ConsumeConcurrentlyStatus.RECONSUME_LATER;

       }

       return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

     }

   });

   consumer.start();

   System.out.printf("Consumer Started.%n");

 }

}

剩余128页未读，继续阅读

岁月玲珑

粉丝: 171
资源: 30

RocketMQ入门与架构解析

TWRP-20190418-3.3.0.0-virgo.img

ASA5500 Series IOS asa912-k8.bin asdm-713.bin

matlab11-note.pdf

NVM-Express-1_4a-2020.03.09-Ratified.pdf

Metaverse-Note 元宇宙笔记metaverse-note.zip

linux-0.11-note.zip

Git-learning-note.txt

telink EMI-test-tool-note.docx

8080-apache-tomcat-7.0.79-note.zip

开源项目-gumieri-note.zip

最新资源