RocketMQ 특성, 전문 용어 (프로듀서, 프로듀서 그룹, 소비자 그룹, 주제, 메시지, 태그, 브로커, 이름 서버) 등

9617 단어 kafkarocketmq (메시지 큐) 
 
  
 
   
RocketMQ 是什么
 
   
Github 上关于 RocketMQ 的介绍： RcoketMQ 是一款低延迟、高可靠、可伸缩、易于使用的消息中间件。具有以下特性：
 
   
     
    
  1. 支持发布/订阅（Pub/Sub）和点对点（P2P）消息模型
    2.  
    
  2. 在一个队列中可靠的先进先出（FIFO）和严格的顺序传递
    3.  
    
  3. 支持拉（pull）和推（push）两种消息模式
    4.  
    
  4. 单一队列百万消息的堆积能力
    5.  
    
  5. 支持多种消息协议，如 JMS、MQTT 等
    6.  
    
  6. 分布式高可用的部署架构,满足至少一次消息传递语义
    7.  
    
  7. 提供 docker 镜像用于隔离测试和云集群部署
    8.  
    
  8. 提供配置、指标和监控等功能丰富的 Dashboard
    9.  
   
 
   
对于这些特性描述，大家简单过一眼就即可，深入学习之后自然就明白了。
 
   
专业术语
 
   
Producer
 
   
消息生产者，生产者的作用就是将消息发送到 MQ，生产者本身既可以产生消息，如读取文本信息等。也可以对外提供接口，由外部应用来调用接口，再由生产者将收到的消息发送到 MQ。
 
   
Producer Group
 
   
生产者组，简单来说就是多个发送同一类消息的生产者称之为一个生产者组。在这里可以不用关心，只要知道有这么一个概念即可。
 
   
Consumer
 
   
消息消费者，简单来说，消费 MQ 上的消息的应用程序就是消费者，至于消息是否进行逻辑处理，还是直接存储到数据库等取决于业务需要。
 
   
Consumer Group
 
   
消费者组，和生产者类似，消费同一类消息的多个 consumer 实例组成一个消费者组。
 
   
Topic
 
   
Topic 是一种消息的逻辑分类，比如说你有订单类的消息，也有库存类的消息，那么就需要进行分类，一个是订单 Topic 存放订单相关的消息，一个是库存 Topic 存储库存相关的消息。
 
   
Message
 
   
Message 是消息的载体。一个 Message 必须指定 topic，相当于寄信的地址。Message 还有一个可选的 tag 设置，以便消费端可以基于 tag 进行过滤消息。也可以添加额外的键值对，例如你需要一个业务 key 来查找 broker 上的消息，方便在开发过程中诊断问题。
 
   
Tag
 
   
标签可以被认为是对 Topic 进一步细化。一般在相同业务模块中通过引入标签来标记不同用途的消息。
 
   
Broker
 
   
Broker 是 RocketMQ 系统的主要角色，其实就是前面一直说的 MQ。Broker 接收来自生产者的消息，储存以及为消费者拉取消息的请求做好准备。
 
   
Name Server
 
   
Name Server 为 producer 和 consumer 提供路由信息。
 
   
RocketMQ 架构
 
   
  
    

      RocketMQ 架构 
    
 
   
 
   
由这张图可以看到有四个集群，分别是 NameServer 集群、Broker 集群、Producer 集群和 Consumer 集群：
 
   
     
    
  1. NameServer: 提供轻量级的服务发现和路由。 每个 NameServer 记录完整的路由信息，提供等效的读写服务，并支持快速存储扩展。
    2.  
    
  2. Broker: 通过提供轻量级的 Topic 和 Queue 机制来处理消息存储,同时支持推（push）和拉（pull）模式以及主从结构的容错机制。
    3.  
    
  3. Producer：生产者，产生消息的实例，拥有相同 Producer Group 的 Producer 组成一个集群。
    4.  
    
  4. Consumer：消费者，接收消息进行消费的实例，拥有相同 Consumer Group 的 Consumer 组成一个集群。
    5.  
   
 
   
简单说明一下图中箭头含义，从 Broker 开始，Broker Master1 和 Broker Slave1 是主从结构，它们之间会进行数据同步，即 Date Sync。同时每个 Broker 与 NameServer 集群中的所有节 点建立长连接，定时注册 Topic 信息到所有 NameServer 中。
 
   
Producer 与 NameServer 集群中的其中一个节点（随机选择）建立长连接，定期从 NameServer 获取 Topic 路由信息，并向提供 Topic 服务的 Broker Master 建立长连接，且定时向 Broker 发送心跳。Producer 只能将消息发送到 Broker master，但是 Consumer 则不一样，它同时和提供 Topic 服务的 Master 和 Slave 建立长连接，既可以从 Broker Master 订阅消息，也可以从 Broker Slave 订阅消息。
 
   
RocketMQ 集群部署模式
 
   
     
    
  1. 单 master 模式 也就是只有一个 master 节点，称不上是集群，一旦这个 master 节点宕机，那么整个服务就不可用，适合个人学习使用。
    2.  
    
  2. 多 master 模式 多个 master 节点组成集群，单个 master 节点宕机或者重启对应用没有影响。 优点：所有模式中性能最高 缺点：单个 master 节点宕机期间，未被消费的消息在节点恢复之前不可用，消息的实时性就受到影响。 注意：使用同步刷盘可以保证消息不丢失，同时 Topic 相对应的 queue 应该分布在集群中各个节点，而不是只在某各节点上，否则，该节点宕机会对订阅该 topic 的应用造成影响。
    3.  
    
  3. 多 master 多 slave 异步复制模式 在多 master 模式的基础上，每个 master 节点都有至少一个对应的 slave。master 节点可读可写，但是 slave 只能读不能写，类似于 mysql 的主备模式。 优点： 在 master 宕机时，消费者可以从 slave 读取消息，消息的实时性不会受影响，性能几乎和多 master 一样。 缺点：使用异步复制的同步方式有可能会有消息丢失的问题。
    4.  
    
  4. 多 master 多 slave 同步双写模式 同多 master 多 slave 异步复制模式类似，区别在于 master 和 slave 之间的数据同步方式。 优点：同步双写的同步模式能保证数据不丢失。 缺点：发送单个消息 RT 会略长，性能相比异步复制低10%左右。 刷盘策略：同步刷盘和异步刷盘（指的是节点自身数据是同步还是异步存储） 同步方式：同步双写和异步复制（指的一组 master 和 slave 之间数据的同步） 注意：要保证数据可靠，需采用同步刷盘和同步双写的方式，但性能会较其他方式低。
    5.  
   
 
  
 
  
 
  
 
   
RocketMQ 单主部署
 
   
鉴于是快速入门，我选择的是第一种单 master 的部署模式。先说明一下我的安装环境：
 
   
     
    
  1. Centos 7.2
    2.  
    
  2. jdk 1.8
    3.  
    
  3. Maven 3.2.x
    4.  
    
  4. Git
    5.  
   
 
   
这里 git 可用可不用，主要是用来直接下载 github 上的源码。也可以选择自己到 github 上下载，然后上传到服务器上。以git操作为示例。
 
   
     
    
  1. clone 源码并用 maven 编译
    2.  
   
 
   
> git clone https://github.com/alibaba/RocketMQ.git /opt/RocketMQ
> cd /opt/RocketMQ && mvn -Dmaven.test.skip=true clean package install assembly:assembly -U
> cd target/alibaba-rocketmq-broker/alibaba-rocketmq

  
여기 서 발생 할 수 있 는 문제 1. "git clone" 을 실행 합 니 다.https://github.com/alibaba/RocketMQ.git / home / inspkgs / RocketMQ "시 다음 과 같은 알림 이 나타 납 니 다.
fatal: unable to access 'https://github.com/alibaba/RocketMQ.git/': Could not resolve host: github.com; Unknown error

  
해결 방법: 일반적으로 네트워크 원인 으로 인 한 것 이 므 로 다음 명령 을 집행 합 니 다.
> ping github.com

  
ping 이 가능 한 지 확인 한 후 git clone 명령 을 다시 실행 합 니 다.2. "mvn - DMaven. test. skip = true clean package install assembly: assembly - U" 컴 파일 을 실행 할 때 관련 jar 를 다운로드 하 는 것 이 느 릴 수 있 습 니 다.이것 도 기본 maven 중앙 창고 가 외국 에 있 기 때문에 필요 에 따라 / home / maven / conf / setting. xml 에 다음 내용 을 추가 한 후 다시 컴 파일 할 수 있 습 니 다.

    aliyun
    central
    aliyun maven
    http://maven.aliyun.com/nexus/content/groups/public/


  
  •  Name Server 시작 
    • > nohup sh /opt/RocketMQ/bin/mqnamesrv &
//   jps     
> jps
25913 NamesrvStartup
//             
> tail -f ~/logs/rocketmqlogs/namesrv.log
The Name Server boot success...

      
  •  시동 브로커 
    • > nohup sh /opt/RocketMQ/bin/mqbroker -n localhost:9876 &
//   jps     
> jps
25954 BrokerStartup
//             
> tail -f ~/logs/rocketmqlogs/broker.log 
The broker[broker-a, 10.1.54.121:10911] boot success...

      
  •  메 시 지 를 보 내 고 받 기 전에 클 라 이언 트 Nameserver 주 소 를 알려 야 합 니 다.RocketMQ 는 이 목 표를 실현 하기 위해 다양한 방식 을 제공 했다.간단하게 보기 위해 환경 변 수 를 사용 합 니 다 NAMESRVADDR。 
    •   
    > export NAMESRV_ADDR=localhost:9876
> sh /opt/RocketMQ/bin/tools.sh com.alibaba.rocketmq.example.quickstart.Producer
SendResult [sendStatus=SEND_OK, msgId= ...
> sh /opt/RocketMQ/bin/tools.sh com.alibaba.rocketmq.example.quickstart.Consumer
ConsumeMessageThread_%d Receive New Messages: [MessageExt...

      
  •  서비스 종료 
    • > sh /opt/RocketMQ/bin/mqshutdown broker
The mqbroker(36695) is running...
Send shutdown request to mqbroker(36695) OK
> sh /opt/RocketMQ/bin/mqshutdown namesrv
The mqnamesrv(36664) is running...
Send shutdown request to mqnamesrv(36664) OK

      
    생산자, 소비자 데모
  •  생산자 
    • public class Producer {
    public static void main(String[] args) throws MQClientException, InterruptedException {

        //        producer
        //    producer group           ，   producer1
        DefaultMQProducer producer = new DefaultMQProducer("producer1");
        
        //  NameServer  ,       NameServer  ，       ；  
        //NameServer      ，                ，           
        producer.setNamesrvAddr("10.1.54.121:9876;10.1.54.122:9876");
        
        //  start()      producer  
        producer.start();

        //  10    Topic TopicTest，tag TagA，     “Hello RocketMQ”   i  
        for (int i = 0; i < 10; i++) {
            try {
                Message msg = new Message("TopicTest",// topic
                        "TagA",// tag
                        ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)// body
                );
                
                //  producer send()      
                //           ，        
                SendResult sendResult = producer.send(msg);
                
                //      ，                   
                System.out.println(sendResult);
            } catch (Exception e) {
                e.printStackTrace();
                Thread.sleep(1000);
            }
        }

        //       ，  shutdown()    producer
        producer.shutdown();
    }
}

      
  •  소비자 
    • public class Consumer {

    public static void main(String[] args) throws InterruptedException, MQClientException {
    
        //        consumer
        //    consumer group           ，   consumer1
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer1");

        //      NameServer  
        consumer.setNamesrvAddr("10.1.54.121:9876;10.1.54.122:9876");

        //        consumer     
        //CONSUME_FROM_LAST_OFFSET     ，          ，       
        //CONSUME_FROM_FIRST_OFFSET           ，     （    broker ）      
        //CONSUME_FROM_TIMESTAMP           ， setConsumeTimestamp()    ，         
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        //  consumer    Topic Tag，*     Tag
        consumer.subscribe("TopicTest", "*");

        //    Listener，           
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List msgs,
                                                            ConsumeConcurrentlyContext context) {

                System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);
                
                //      
                //CONSUME_SUCCESS     
                //RECONSUME_LATER     ，        
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        //  start()    consumer
        consumer.start();

        System.out.println("Consumer Started.");
    }
}

      
    저자: 풍 선생 의 노트
    링크:http://www.jianshu.com/p/824066d70da8 
    약서
    저작권 은 작가 에 게 있다.상업 전 재 는 작가 에 게 연락 하여 권한 을 수 여 받 으 십시오. 비 상업 전 재 는 출처 를 밝 혀 주 십시오.저자: 풍 선생 의 필기 링크:http://www.jianshu.com/p/824066d70da8출처: 약서 저작권 은 작가 소유.상업 전 재 는 작가 에 게 연락 하여 권한 을 수 여 받 으 십시오. 비 상업 전 재 는 출처 를 밝 혀 주 십시오.
    arcgis api for js 동적 그래 픽 트 리 생 성 및 제어
    자바 정적 검색 도구 에 대한 간단 한 소개

    좋은 웹페이지 즐겨찾기

    개발자 우수 사이트 수집

    개발자가 알아야 할 필수 사이트 100선 추천 우리는 당신을 위해 100개의 자주 사용하는 개발자 학습 사이트를 정리했습니다
    best100-homepage