RocketMQ（01）——简介

短命女 2023-06-20 06:09 53阅读 0赞

# RocketMQ简介 #

笔者使用的是Apache RocketMQ，官网是[http://rocketmq.apache.org/][http_rocketmq.apache.org]。RocketMQ是Alibaba开源的一个分布式消息队列，可以通过[http://rocketmq.apache.org/dowloading/releases/][http_rocketmq.apache.org_dowloading_releases]下载当前最新的版本。下载后解压缩，然后通过`bin/mqnamesrv`启动一个Name Server，它默认监听在9876端口。然后需要通过`bin/mqbroker -n localhost:9876`启动一个Broker，并把它注册到刚刚启动的那个Name Server上，Broker默认监听在端口10911上。生产者和消费者都是跟Broker打交道，但是它们不会直接指定Broker的地址，而是通过Name Server来间接的获取Broker的地址。这样做的好处是可以动态的增加Broker，多个Broker之间可以组成一个集群。应用中使用RocketMQ时需要添加RocketMQ Client依赖。

<dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-client</artifactId>
        <version>4.5.0</version>
    </dependency>

然后可以通过DefaultMQProducer进行消息的发送，每一个生产者必须指定一个Group，下面代码就指定了Group为group1。相同处理逻辑的生产者必须定义相同的Group。这个Group只对于发送事务消息的生产者有用。然后需要通过`setNamesrvAddr()`指定Name Server的地址。在发送消息前必须调用其`start()`。发送的消息是通过`org.apache.rocketmq.common.message.Message`对象表示的。它必须要指定一个Topic，RocketMQ是通过抽象的Topic来管理一组队列的，这个Topic必须在Broker中进行创建。可以通过`bin/mqadmin updateTopic -b localhost:10911 -t topic1`在本地刚刚启动的Broker上创建名为topic1的Topic。它默认拥有8个读队列，8个写队列。下面的代码指定了消息都将发送到名为topic1的Topic。通过其`send()`进行消息发送，它是同步发送的，发送完后会返回一个SendResult。其SendStatus为SEND\_OK时表示发送成功。下面的代码一共发送了10条消息到topic1,消息内容分别是hello0…hello9。

@Test
    public void testSend() throws Exception { 
      //指定Producer的Group为group1
      DefaultMQProducer producer = new DefaultMQProducer("group1");
      //指定需要连接的Name Server
      producer.setNamesrvAddr(nameServer);
      //发送消息前必须调用start()，其内部会进行一些初始化工作。
      producer.start();
      for (int i = 0; i < 10; i++) { 
        //指定消息发送的Topic是topic1。
        Message message = new Message("topic1", ("hello" + i).getBytes());
        //同步发送，发送成功后才会返回
        SendResult sendResult = producer.send(message);
        if (sendResult.getSendStatus() == SendStatus.SEND_OK) { 
          System.out.println("消息发送成功：" + sendResult);
        } else { 
          System.out.println("消息发送失败：" + sendResult);
        }
      }
      //使用完毕后需要把Producer关闭，以释放相应的资源
      producer.shutdown();
    }

消息的消费者可以通过DefaultMQPushConsumer进行消费。DefaultMQPushConsumer是进行推模式消费的，它也需要指定一个Group。默认情况下相同Group的消费者将对同一个队列中的消息进行集群消费，即同一条消息只会被一个Consumer实例进行消费。DefaultMQPushConsumer也需要通过`setNamesrvAddr()`指定需要连接的Name Server。通过`subscribe()`指定需要消费的Topic和对应的Tag。下面指定了需要消费的Topic是topic1，通过`*`指定将消费所有的Tag。Tag是用来对消息进行分类标记的，需要在发送消息的时候指定。通过`registerMessageListener()`注册消息监听器，当收到消息后会回调它。下面代码注册的是一个MessageListenerConcurrently类型的监听器。消息正常消费后需要返回CONSUME\_SUCCESS，如果消费失败可以返回RECONSUME\_LATER，这样可以先跳过这一条消息的消费，Broker会过一段时间再投递这一条消息。Consumer也是需要通过`start()`进行启动。这样消费者就可以开始进行消息消费了，默认只有它启动之后发送的消息才能收到。

@Test
    public void testConsumer() throws Exception { 
      //创建Consumer并指定消费者组。
      DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group1");
      //指定需要连接的Name Server
      consumer.setNamesrvAddr(nameServer);
      //订阅topic1上的所有Tag。
      consumer.subscribe("topic1", "*");
      //注册一个消息监听器
      consumer.registerMessageListener(new MessageListenerConcurrently() { 
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { 
          System.out.println(Thread.currentThread().getName() + "收到了消息，数量是：" + msgs.size());
          AtomicInteger counter = new AtomicInteger();
          msgs.forEach(msg -> System.out.println(counter.incrementAndGet() + ".消息内容是：" + new String(msg.getBody())));
          return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
      });
      //启动消费者
      consumer.start();
      //为了确保Junit线程不立即死掉。
      TimeUnit.SECONDS.sleep(120);
    }

（注：本文是基于Apache RocketMQ4.5.0所写）

[http_rocketmq.apache.org]: http://rocketmq.apache.org/
[http_rocketmq.apache.org_dowloading_releases]: http://rocketmq.apache.org/dowloading/releases/